wren
85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
"""企查查企业信息缓存模型。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import BigInteger, Index, String, or_, select
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from fastapi_common.fastapi_common_web.models import BaseModel
|
|
|
|
|
|
class QichachaCompanyInfo(BaseModel):
|
|
"""企查查企业信息缓存表。"""
|
|
|
|
__tablename__ = "qcc_company_info"
|
|
__table_args__ = (
|
|
Index("idx_qcc_company_search_key", "search_key", unique=True),
|
|
Index("idx_qcc_company_credit_code", "credit_code"),
|
|
Index("idx_qcc_company_name", "company_name"),
|
|
)
|
|
|
|
Id: Mapped[int] = mapped_column("id", BigInteger, primary_key=True, autoincrement=True)
|
|
searchKey: Mapped[str] = mapped_column("search_key", String(200), nullable=False, comment="查询关键词")
|
|
creditCode: Mapped[str | None] = mapped_column("credit_code", String(64), comment="统一社会信用代码")
|
|
companyName: Mapped[str | None] = mapped_column("company_name", String(255), comment="企业名称")
|
|
enterprise: Mapped[dict[str, Any] | None] = mapped_column("enterprise", JSONB, comment="工商信息")
|
|
dishonesty: Mapped[dict[str, Any] | None] = mapped_column("dishonesty", JSONB, comment="失信信息")
|
|
|
|
@classmethod
|
|
async def FindByKeyword(cls, session: AsyncSession, Keyword: str) -> "QichachaCompanyInfo | None":
|
|
"""按查询词、信用代码或企业名查询缓存。"""
|
|
return await session.scalar(
|
|
select(cls).where(
|
|
cls.deleted_at.is_(None),
|
|
or_(
|
|
cls.searchKey == Keyword,
|
|
cls.creditCode == Keyword,
|
|
cls.companyName == Keyword,
|
|
),
|
|
)
|
|
)
|
|
|
|
@classmethod
|
|
async def Upsert(
|
|
cls,
|
|
session: AsyncSession,
|
|
SearchKey: str,
|
|
CreditCode: str | None,
|
|
CompanyName: str | None,
|
|
Enterprise: dict[str, Any] | None,
|
|
Dishonesty: dict[str, Any] | None,
|
|
) -> "QichachaCompanyInfo":
|
|
"""按查询关键词写入或更新缓存。"""
|
|
record = await cls.FindByKeyword(session, SearchKey)
|
|
if record is None:
|
|
record = cls(
|
|
searchKey=SearchKey,
|
|
creditCode=CreditCode,
|
|
companyName=CompanyName,
|
|
enterprise=Enterprise,
|
|
dishonesty=Dishonesty,
|
|
)
|
|
session.add(record)
|
|
else:
|
|
record.searchKey = SearchKey
|
|
record.creditCode = CreditCode
|
|
record.companyName = CompanyName
|
|
record.enterprise = Enterprise
|
|
record.dishonesty = Dishonesty
|
|
record.updated_at = datetime.now(UTC)
|
|
await session.flush()
|
|
return record
|
|
|
|
@classmethod
|
|
def GetAgeDays(cls, record: "QichachaCompanyInfo") -> int:
|
|
"""计算缓存记录距今天数。"""
|
|
updated_at = record.updated_at
|
|
if updated_at.tzinfo is None:
|
|
updated_at = updated_at.replace(tzinfo=UTC)
|
|
return max((datetime.now(UTC) - updated_at).days, 0)
|