conference-archive/confarchive/model.py

321 lines
9.3 KiB
Python

"""Database models."""
import os
import typing
import sqlalchemy
import sqlalchemy.orm.decl_api
from sqlalchemy import func
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.orderinglist import ordering_list
from sqlalchemy.orm import relationship
from sqlalchemy.schema import Column, ForeignKey
from sqlalchemy.types import Boolean, Date, DateTime, Integer, String
from .database import session
Base: sqlalchemy.orm.decl_api.DeclarativeMeta = declarative_base()
Base.query = session.query_property()
content_type_to_extension = {
"image/jpeg": "jpg",
"image/png": "png",
"image/gif": "gif",
}
class TimeStampedModel(Base):
"""Time stamped model."""
__abstract__ = True
created = Column(DateTime, default=func.now())
modified = Column(DateTime, default=func.now(), onupdate=func.now())
class Series(TimeStampedModel):
"""Conference series."""
__tablename__ = "series"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
slug = Column(String, unique=True)
wikidata_qid = Column(String, unique=True)
conferences = relationship(
"Conference",
order_by="Conference.start",
back_populates="series",
)
class Conference(TimeStampedModel):
"""Conference."""
__tablename__ = "conference"
id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
start = Column(Date)
end = Column(Date)
days = Column(Integer)
timezone = Column(String)
location = Column(String)
country = Column(String)
acronym = Column(String)
url = Column(String)
schedule_xml_url = Column(String)
short_name = Column(String, unique=True)
venue_id = Column(Integer, ForeignKey("venue.id"))
online = Column(Boolean)
wikidata_qid = Column(String, unique=True)
series_id = Column(Integer, ForeignKey("series.id"))
people_detail = relationship(
"ConferencePerson", lazy="dynamic", back_populates="conference"
)
people = association_proxy("people_detail", "person")
events = relationship(
"Event",
order_by="Event.event_date",
back_populates="conference",
lazy="dynamic",
)
venue = relationship("Venue", back_populates="conferences")
series = relationship("Series", back_populates="conferences")
class City(TimeStampedModel):
"""City."""
__tablename__ = "city"
id = Column(Integer, primary_key=True)
slug = Column(String, nullable=False, unique=True)
name = Column(String, nullable=False)
country_code = Column(String, ForeignKey("country.alpha2"), nullable=False)
wikidata_qid = Column(String, nullable=False)
venues = relationship("Venue", back_populates="city")
country = relationship("Country", back_populates="cities")
class Venue(TimeStampedModel):
"""Venue."""
__tablename__ = "venue"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
city_id = Column(Integer, ForeignKey("city.id"))
wikidata_qid = Column(String, nullable=False)
conferences = relationship("Conference", back_populates="venue")
city = relationship("City", back_populates="venues")
class Country(TimeStampedModel):
"""Country."""
__tablename__ = "country"
alpha2 = Column(String, primary_key=True)
name = Column(String)
wikidata_qid = Column(String, nullable=False)
cities = relationship("City", back_populates="country")
@property
def flag(self) -> str:
a = ord("A")
flag_a = 0x1F1E6
char1, char2 = (
flag_a + ord(self.alpha2[0]) - a,
flag_a + ord(self.alpha2[1]) - a,
)
return chr(char1) + chr(char2)
class ConferencePerson(Base):
__tablename__ = "conference_person"
conference_id = Column(Integer, ForeignKey("conference.id"), primary_key=True)
person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
named_as = Column(String)
bio = Column(String)
slug = Column(String)
url = Column(String)
affiliation = Column(String)
photo_url = Column(String)
photo_url_content_type = Column(String)
person = relationship("Person", back_populates="conferences_association")
conference = relationship("Conference", back_populates="people_detail")
@property
def events(self):
return (
Event.query.join(EventPerson)
.filter(
Event.conference == self.conference, EventPerson.person == self.person
)
.order_by(Event.event_date.desc())
)
class Event(TimeStampedModel):
"""Event."""
__tablename__ = "event"
id = Column(Integer, primary_key=True)
conference_id = Column(Integer, ForeignKey("conference.id"), nullable=False)
event_date = Column(DateTime)
# day = Column(Integer)
guid = Column(String)
start = Column(String)
duration = Column(String)
room = Column(String)
track = Column(String)
slug = Column(String)
title = Column(String, nullable=False)
abstract = Column(String)
description = Column(String)
event_type = Column(String)
url = Column(String)
cancelled = Column(Boolean)
conference = relationship("Conference", back_populates="events")
people_detail = relationship(
"EventPerson",
order_by="EventPerson.position",
back_populates="event",
collection_class=ordering_list("position"),
)
people = association_proxy(
"people_detail",
"person",
creator=lambda i: EventPerson(person=i),
)
class Person(TimeStampedModel):
"""Person."""
__tablename__ = "person"
id = Column(Integer, primary_key=True)
name = Column(String)
wikidata_qid = Column(String)
gender = Column(String)
wikidata_photo = Column(postgresql.ARRAY(String))
events_association = relationship(
"EventPerson",
back_populates="person",
lazy="dynamic",
)
events = association_proxy("events_association", "event")
conferences_association = relationship(
"ConferencePerson", lazy="dynamic", back_populates="person"
)
conferences = association_proxy("conferences_association", "conference")
@property
def conference_count(self):
return ConferencePerson.query.filter_by(person_id=self.id).count()
@property
def event_count(self):
return EventPerson.query.filter_by(person_id=self.id).count()
def active_years(self):
q = (
session.query(func.min(Event.event_date), func.max(Event.event_date))
.join(EventPerson)
.filter_by(person_id=self.id)
)
return q.one()
# photos = relationship("PersonPhoto", back_populates="person")
def events_by_time(self):
q = (
session.query(Event)
.join(EventPerson)
.filter(EventPerson.person == self)
.order_by(Event.event_date.desc())
)
return q
def conference_by_time(self):
q = (
session.query(ConferencePerson)
.join(Conference)
.filter(ConferencePerson.person == self)
.order_by(Conference.start.desc())
)
return q
def bio_source(self) -> ConferencePerson | None:
bio_list = [cp for cp in self.conferences_association if cp.bio]
if not bio_list:
return None
if len(bio_list) == 1:
return typing.cast(ConferencePerson, bio_list[0])
recent = max(bio_list, key=lambda cp: cp.conference.start)
len_recent_bio = len(recent.bio)
longest = max(bio_list, key=lambda cp: len(cp.bio))
if recent == longest:
return typing.cast(ConferencePerson, recent)
best = longest if len(longest.bio) > len_recent_bio * 2 else recent
return typing.cast(ConferencePerson, best)
def photo_filename(self) -> str | None:
"""Speaker photo filename."""
if self.wikidata_photo:
assert isinstance(self.wikidata_photo[0], str)
return os.path.join("wikidata_photo", "thumb", self.wikidata_photo[0])
q = self.conferences_association.filter(
ConferencePerson.photo_url.isnot(None),
ConferencePerson.photo_url_content_type.isnot(None),
)
if q.count() == 0:
return None
best = max(q, key=lambda cp: cp.conference.start)
ext = content_type_to_extension[best.photo_url_content_type]
filename = f"{best.conference_id}_{self.id}.{ext}"
return os.path.join("conference_photo", filename)
# class PersonPhoto(TimeStampedModel):
# """Person photo."""
#
# __tablename__ = "person_photo"
# person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
# source_filename = Column(String)
#
# person = relationship("Person", back_populates="photos")
class EventPerson(Base):
"""Event person."""
__tablename__ = "event_person"
event_id = Column(Integer, ForeignKey("event.id"), primary_key=True)
person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
position = Column(Integer, nullable=False)
person = relationship("Person", back_populates="events_association")
event = relationship("Event", back_populates="people_detail")