feat: relation between person and character

pol/api/v0/character.py

@@ -11,10 +11,17 @@
 from pol.router import ErrorCatchRoute
 from pol.depends import get_db, get_redis
 from pol.db.const import Gender, get_character_rel
-from pol.db.tables import ChiiCharacter, ChiiPersonField, ChiiCrtSubjectIndex
+from pol.db.tables import (
+ ChiiPerson,
+ ChiiSubject,
+ ChiiCharacter,
+ ChiiPersonField,
+ ChiiCrtCastIndex,
+ ChiiCrtSubjectIndex,
+)
 from pol.api.v0.const import NotFoundDescription
 from pol.api.v0.utils import person_images
-from pol.api.v0.models import RelatedSubject, CharacterDetail
+from pol.api.v0.models import RelatedSubject, CharacterDetail, CharacterPerson
 from pol.redis.json_cache import JSONRedis
 
 router = APIRouter(tags=["角色"], route_class=ErrorCatchRoute)
@@ -143,3 +150,61 @@ async def get_person_subjects(
  )
 
  return subjects
+
+
+@router.get(
+ "/characters/{character_id}/persons",
+ summary="get character related persons",
+ response_model=List[CharacterPerson],
+ responses={
+ 404: res.response(model=ErrorDetail),
+ },
+)
+async def get_character_persons(
+ db: AsyncSession = Depends(get_db),
+ not_found: Exception = Depends(exc_404),
+ character_id: int = Path(..., gt=0),
+):
+ character: Optional[ChiiCharacter] = await db.scalar(
+ sa.select(ChiiCharacter)
+ .where(ChiiCharacter.crt_id == character_id, ChiiCharacter.crt_ban == 0)
+ .limit(1)
+ )
+
+ if character is None:
+ raise not_found
+
+ query = (
+ sa.select(
+ ChiiCrtCastIndex.crt_id,
+ ChiiCrtCastIndex.prsn_id,
+ ChiiPerson.prsn_name,
+ ChiiPerson.prsn_type,
+ ChiiPerson.prsn_img,
+ ChiiSubject.subject_id,
+ ChiiSubject.subject_name,
+ ChiiSubject.subject_name_cn,
+ )
+ .distinct()
+ .join(ChiiPerson, ChiiPerson.prsn_id == ChiiCrtCastIndex.prsn_id)
+ .join(ChiiSubject, ChiiSubject.subject_id == ChiiCrtCastIndex.subject_id)
+ .where(
+ ChiiCrtCastIndex.crt_id == character.crt_id,
+ ChiiPerson.prsn_ban == 0,
+ )
+ )
+
+ persons = [
+ {
+ "id": r["prsn_id"],
+ "name": r["prsn_name"],
+ "type": r["prsn_type"],
+ "images": person_images(r["prsn_img"]),
+ "subject_id": r["subject_id"],
+ "subject_name": r["subject_name"],
+ "subject_name_cn": r["subject_name_cn"],
+ }
+ for r in (await db.execute(query)).mappings().fetchall()
+ ]
+
+ return persons

pol/api/v0/models/__init__.py

@@ -21,6 +21,8 @@
  "RelatedSubject",
  "Character",
  "CharacterDetail",
+ "CharacterPerson",
+ "PersonCharacter",
 ]
 
 
@@ -151,6 +153,18 @@ class CharacterDetail(BaseCharacter):
  stat: Stat
 
 
+class PersonCharacter(BaseCharacter):
+ subject_id: int
+ subject_name: str
+ subject_name_cn: str
+
+
+class CharacterPerson(BaseCharacter):
+ subject_id: int
+ subject_name: str
+ subject_name_cn: str
+
+
 class Pager(pydantic.BaseModel):
  limit: int = Field(30, gt=0, le=50)
  offset: int = Field(0, ge=0)

pol/api/v0/person.py

@@ -11,10 +11,17 @@
 from pol.router import ErrorCatchRoute
 from pol.depends import get_db, get_redis
 from pol.db.const import Gender, StaffMap
-from pol.db.tables import ChiiPerson, ChiiPersonCsIndex, ChiiCharacterField
+from pol.db.tables import (
+ ChiiPerson,
+ ChiiSubject,
+ ChiiCharacter,
+ ChiiCrtCastIndex,
+ ChiiPersonCsIndex,
+ ChiiCharacterField,
+)
 from pol.api.v0.const import NotFoundDescription
 from pol.api.v0.utils import get_career, person_images
-from pol.api.v0.models import PersonDetail, RelatedSubject
+from pol.api.v0.models import PersonDetail, RelatedSubject, PersonCharacter
 from pol.redis.json_cache import JSONRedis
 
 router = APIRouter(tags=["人物"], route_class=ErrorCatchRoute)
@@ -148,6 +155,64 @@ async def get_person_subjects(
  return subjects
 
 
+@router.get(
+ "/persons/{person_id}/characters",
+ summary="get person related characters",
+ response_model=List[PersonCharacter],
+ responses={
+ 404: res.response(model=ErrorDetail),
+ },
+)
+async def get_person_characters(
+ db: AsyncSession = Depends(get_db),
+ not_found: Exception = Depends(exc_404),
+ person_id: int = Path(..., gt=0),
+):
+ person: Optional[ChiiPerson] = await db.scalar(
+ sa.select(ChiiPerson)
+ .where(ChiiPerson.prsn_id == person_id, ChiiPerson.prsn_ban == 0)
+ .limit(1)
+ )
+ if person is None:
+ raise not_found
+
+ query = (
+ sa.select(
+ ChiiCrtCastIndex.crt_id,
+ ChiiCrtCastIndex.prsn_id,
+ ChiiCharacter.crt_name,
+ ChiiCharacter.crt_role,
+ ChiiCharacter.crt_img,
+ ChiiSubject.subject_id,
+ ChiiSubject.subject_name,
+ ChiiSubject.subject_name_cn,
+ )
+ .distinct()
+ .join(ChiiCharacter, ChiiCharacter.crt_id == ChiiCrtCastIndex.crt_id)
+ .join(ChiiSubject, ChiiSubject.subject_id == ChiiCrtCastIndex.subject_id)
+ .where(
+ ChiiCrtCastIndex.prsn_id == person.prsn_id,
+ ChiiCharacter.crt_redirect == 0,
+ ChiiCharacter.crt_ban == 0,
+ )
+ )
+
+ characters = [
+ {
+ "id": r["crt_id"],
+ "name": r["crt_name"],
+ "type": r["crt_role"],
+ "images": person_images(r["crt_img"]),
+ "subject_id": r["subject_id"],
+ "subject_name": r["subject_name"],
+ "subject_name_cn": r["subject_name_cn"],
+ }
+ for r in (await db.execute(query)).mappings().fetchall()
+ ]
+
+ return characters
+
+
 def person_img_url(s: Optional[str]) -> Optional[str]:
  if not s:
  return None

pol/db/const.py

@@ -16,12 +16,18 @@ class BloodType(enum.IntEnum):
  ab = 3
  o = 4
 
+ def translate(self, _escape_table):
+ return self.value
+
 
 class PersonType(enum.IntEnum):
  person = 1
  company = 2
  band = 3
 
+ def translate(self, _escape_table):
+ return self.value
+
 
 class Gender(enum.IntEnum):
  male = 1
@@ -34,6 +40,9 @@ def str(self):
  return "female"
  raise ValueError(f"{self.value} is not valid gender")
 
+ def translate(self, _escape_table):
+ return self.value
+
 
 class EpType(enum.IntEnum):
  normal = 0

tests/app/api_v0/test_character.py

@@ -108,3 +108,22 @@ def test_character_lock(client: TestClient):
 
  res = response.json()
  assert res["locked"]
+
+
+def test_character_persons(client: TestClient, mock_person):
+ mock_person(3818, "福山潤")
+ response = client.get("/v0/characters/1/persons")
+ assert response.status_code == 200
+
+ persons = response.json()
+ assert persons[0]["id"] == 3818
+ assert persons[0]["subject_id"] == 8
+ assert set(persons[0].keys()) == {
+ "id",
+ "name",
+ "type",
+ "images",
+ "subject_id",
+ "subject_name",
+ "subject_name_cn",
+ }

tests/app/api_v0/test_person.py

@@ -93,3 +93,22 @@ def test_person_lock(client: TestClient):
 
  res = response.json()
  assert res["locked"]
+
+
+def test_person_characters(client: TestClient, mock_person):
+ mock_person(3818, "福山潤")
+ response = client.get("/v0/persons/3818/characters")
+ assert response.status_code == 200
+
+ characters = response.json()
+ assert characters[0]["id"] == 1
+ assert characters[0]["subject_id"] == 8
+ assert set(characters[0].keys()) == {
+ "id",
+ "name",
+ "type",
+ "images",
+ "subject_id",
+ "subject_name",
+ "subject_name_cn",
+ }

tests/conftest.py

@@ -7,10 +7,12 @@
 from sqlalchemy.orm import sessionmaker
 
 from pol import sa, config
-from pol.db.const import SubjectType
+from pol.db.const import Gender, BloodType, PersonType, SubjectType
 from pol.db.tables import (
  ChiiMember,
+ ChiiPerson,
  ChiiSubject,
+ ChiiPersonField,
  ChiiSubjectField,
  ChiiOauthAccessToken,
 )
@@ -122,6 +124,95 @@ def mock_id(
  db_session.commit()
 
 
+@pytest.fixture()
+def mock_person(db_session: Session):
+ mock_person_id = set()
+ delete_query = defaultdict(list)
+
+ def mock_id(
+ prsn_id: int,
+ prsn_name="",
+ prsn_type=PersonType.person,
+ prsn_infobox="",
+ prsn_producer=0,
+ prsn_mangaka=0,
+ prsn_artist=1,
+ prsn_seiyu=0,
+ prsn_writer=0,
+ prsn_illustrator=0,
+ prsn_actor=0,
+ prsn_summary="",
+ prsn_img="",
+ prsn_img_anidb="",
+ prsn_comment=0,
+ prsn_collects=0,
+ prsn_dateline=0,
+ prsn_lastpost=0,
+ prsn_lock=0,
+ prsn_anidb_id=0,
+ prsn_ban=0,
+ prsn_redirect=0,
+ prsn_nsfw=0,
+ gender=Gender.male,
+ bloodtype=BloodType.o,
+ birth_year=2000,
+ birth_mon=1,
+ birth_day=1,
+ ):
+ delete_query[ChiiPerson].append(ChiiPerson.prsn_id == prsn_id)
+ delete_query[ChiiPersonField].append(ChiiPersonField.prsn_id == prsn_id)
+
+ check_exist(db_session, delete_query)
+ mock_person_id.add(prsn_id)
+
+ db_session.add(
+ ChiiPerson(
+ prsn_id=prsn_id,
+ prsn_name=prsn_name,
+ prsn_type=prsn_type,
+ prsn_infobox=prsn_infobox,
+ prsn_producer=prsn_producer,
+ prsn_mangaka=prsn_mangaka,
+ prsn_artist=prsn_artist,
+ prsn_seiyu=prsn_seiyu,
+ prsn_writer=prsn_writer,
+ prsn_illustrator=prsn_illustrator,
+ prsn_actor=prsn_actor,
+ prsn_summary=prsn_summary,
+ prsn_img=prsn_img,
+ prsn_img_anidb=prsn_img_anidb,
+ prsn_comment=prsn_comment,
+ prsn_collects=prsn_collects,
+ prsn_dateline=prsn_dateline,
+ prsn_lastpost=prsn_lastpost,
+ prsn_lock=prsn_lock,
+ prsn_anidb_id=prsn_anidb_id,
+ prsn_ban=prsn_ban,
+ prsn_redirect=prsn_redirect,
+ prsn_nsfw=prsn_nsfw,
+ )
+ )
+ db_session.add(
+ ChiiPersonField(
+ prsn_cat="prsn",
+ prsn_id=prsn_id,
+ gender=gender,
+ bloodtype=bloodtype,
+ birth_year=birth_year,
+ birth_mon=birth_mon,
+ birth_day=birth_day,
+ )
+ )
+ db_session.commit()
+
+ try:
+ yield mock_id
+ finally:
+ for table, where in delete_query.items():
+ db_session.execute(sa.delete(table).where(sa.or_(*where)))
+ db_session.commit()
+
+
 @pytest.fixture()
 def mock_access_token(db_session: Session):
  mock_user_id = set()