141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
"""报告备案：工作人员登记，公众按编号对照。"""
|
|
|
|
from datetime import date
|
|
|
|
from sqlalchemy import Boolean, Date, Integer, String, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.db import Base
|
|
|
|
FOUND_NOTICE = "以出具机构数据为准，协会结果仅供参考。"
|
|
MISSING_NOTICE = "协会备案库没有该编号，请向出具机构核实。"
|
|
|
|
|
|
def _mysql_args() -> dict:
|
|
return {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}
|
|
|
|
|
|
class ReportFiling(Base):
|
|
__tablename__ = "report_filing"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
institution_name: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
report_number: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
|
|
issued_on: Mapped[date] = mapped_column(Date, nullable=False)
|
|
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
|
|
|
|
def _institution(value: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError("请填写出具机构")
|
|
text = value.strip()
|
|
if len(text) > 200:
|
|
raise ValueError("出具机构过长")
|
|
return text
|
|
|
|
|
|
def _number(value: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError("请填写报告编号")
|
|
text = value.strip()
|
|
if len(text) > 64:
|
|
raise ValueError("报告编号过长")
|
|
return text
|
|
|
|
|
|
def _issued_on(value: str) -> date:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError("请填写出具日期")
|
|
try:
|
|
return date.fromisoformat(value.strip())
|
|
except ValueError as exc:
|
|
raise ValueError("出具日期不正确") from exc
|
|
|
|
|
|
def _admin_filing(row: ReportFiling) -> dict:
|
|
return {
|
|
"id": row.id,
|
|
"institutionName": row.institution_name,
|
|
"reportNumber": row.report_number,
|
|
"issuedOn": row.issued_on.isoformat(),
|
|
"published": bool(row.published),
|
|
}
|
|
|
|
|
|
def list_filings(db) -> dict:
|
|
rows = db.scalars(select(ReportFiling).order_by(ReportFiling.id.desc())).all()
|
|
return {"items": [_admin_filing(row) for row in rows]}
|
|
|
|
|
|
def create_filing(db, *, institution_name: str, report_number: str, issued_on: str, published: bool) -> int:
|
|
row = ReportFiling(
|
|
institution_name=_institution(institution_name),
|
|
report_number=_number(report_number),
|
|
issued_on=_issued_on(issued_on),
|
|
published=bool(published),
|
|
)
|
|
db.add(row)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError as exc:
|
|
db.rollback()
|
|
raise ValueError("报告编号已存在") from exc
|
|
return row.id
|
|
|
|
|
|
def update_filing(
|
|
db,
|
|
filing_id: int,
|
|
*,
|
|
institution_name: str,
|
|
report_number: str,
|
|
issued_on: str,
|
|
published: bool,
|
|
) -> None:
|
|
row = db.get(ReportFiling, filing_id)
|
|
if row is None:
|
|
raise ValueError("备案不存在")
|
|
number = _number(report_number)
|
|
taken = db.scalar(select(ReportFiling.id).where(ReportFiling.report_number == number, ReportFiling.id != row.id))
|
|
if taken is not None:
|
|
raise ValueError("报告编号已存在")
|
|
row.institution_name = _institution(institution_name)
|
|
row.report_number = number
|
|
row.issued_on = _issued_on(issued_on)
|
|
row.published = bool(published)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError as exc:
|
|
db.rollback()
|
|
raise ValueError("报告编号已存在") from exc
|
|
|
|
|
|
def delete_filing(db, filing_id: int) -> None:
|
|
row = db.get(ReportFiling, filing_id)
|
|
if row is None:
|
|
raise ValueError("备案不存在")
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
def query_filing(db, number: str) -> dict:
|
|
if not isinstance(number, str) or not number.strip():
|
|
raise ValueError("请填写报告编号")
|
|
text = number.strip()
|
|
if len(text) > 64:
|
|
return {"found": False, "notice": MISSING_NOTICE}
|
|
row = db.scalar(
|
|
select(ReportFiling).where(ReportFiling.report_number == text, ReportFiling.published.is_(True))
|
|
)
|
|
if row is None:
|
|
return {"found": False, "notice": MISSING_NOTICE}
|
|
return {
|
|
"found": True,
|
|
"institutionName": row.institution_name,
|
|
"reportNumber": row.report_number,
|
|
"issuedOn": row.issued_on.isoformat(),
|
|
"notice": FOUND_NOTICE,
|
|
}
|