Spaces:

jun2024
/

fc

Paused

fc / apps /automation /services /document_delivery /data_classes.py

jun2024

Upload 2241 files

b4f2238 verified 7 months ago

8.4 kB

"""
文书送达数据类
定义文书送达相关的数据传输对象
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

# ============================================================
# API 响应数据类 (来自法院一张网 API)
# ============================================================


@dataclass
class DocumentRecord:
    """
    单个文书记录 - 来自 getSdListByZjhmAndAhdmNew API
    对应 Requirements 1.2, 3.1
    """

    ah: str  # 案号，如"（2025）粤0604民初41257号"
    sdbh: str  # 送达编号 - 用于获取文书详情
    ajzybh: str  # 案件主要编号
    fssj: str  # 发送时间，如"2025-12-10 16:25:37" - 用于时间过滤
    fymc: str  # 法院名称
    ahdm: str = ""  # 案号代码
    fybh: str = ""  # 法院编号
    ssdrxm: str = ""  # 送达人姓名
    ssdrsjhm: str = ""  # 送达人手机号
    ssdrzjhm: str = ""  # 送达人证件号码
    wsmc: str = ""  # 文书名称（多个用逗号分隔）
    sdzt: str = ""  # 送达状态
    qdzt: str = ""  # 签到状态
    qdbh: str = ""  # 签到编号
    fqr: str = ""  # 发起人
    cjsj: str = ""  # 创建时间
    zhxgsj: str = ""  # 最后修改时间

    def parse_fssj(self) -> datetime | None:
        """
        解析 fssj（发送时间）字符串为 datetime 对象
        Returns:
            datetime 对象，解析失败返回 None
        """
        if not self.fssj:
            return None
        try:
            return datetime.strptime(self.fssj, "%Y-%m-%d %H:%M:%S")
        except ValueError:
            # 尝试其他格式
            try:
                return datetime.fromisoformat(self.fssj)
            except ValueError:
                return None

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "DocumentRecord":
        """从 API 响应字典创建实例"""
        return cls(
            ah=data.get("ah", ""),
            sdbh=data.get("sdbh", ""),
            ajzybh=data.get("ajzybh", ""),
            fssj=data.get("fssj", ""),
            fymc=data.get("fymc", ""),
            ahdm=data.get("ahdm", ""),
            fybh=data.get("fybh", ""),
            ssdrxm=data.get("ssdrxm", ""),
            ssdrsjhm=data.get("ssdrsjhm", ""),
            ssdrzjhm=data.get("ssdrzjhm", ""),
            wsmc=data.get("wsmc", ""),
            sdzt=data.get("sdzt", ""),
            qdzt=data.get("qdzt", ""),
            qdbh=data.get("qdbh", ""),
            fqr=data.get("fqr", ""),
            cjsj=data.get("cjsj", ""),
            zhxgsj=data.get("zhxgsj", ""),
        )

    def to_dict(self) -> dict[str, Any]:
        """序列化为字典"""
        return {
            "ah": self.ah,
            "sdbh": self.sdbh,
            "ajzybh": self.ajzybh,
            "fssj": self.fssj,
            "fymc": self.fymc,
            "ahdm": self.ahdm,
            "fybh": self.fybh,
            "ssdrxm": self.ssdrxm,
            "ssdrsjhm": self.ssdrsjhm,
            "ssdrzjhm": self.ssdrzjhm,
            "wsmc": self.wsmc,
            "sdzt": self.sdzt,
            "qdzt": self.qdzt,
            "qdbh": self.qdbh,
            "fqr": self.fqr,
            "cjsj": self.cjsj,
            "zhxgsj": self.zhxgsj,
        }


@dataclass
class DocumentDetail:
    """
    文书详情（下载信息）- 来自 getWsListBySdbhNew API
    对应 Requirements 2.2
    """

    c_sdbh: str  # 送达编号
    c_wsmc: str  # 文书名称
    c_wjgs: str  # 文件格式（如 pdf）
    wjlj: str  # 文件链接（OSS URL，带签名）
    c_stbh: str = ""  # 上传编号（文件路径）
    c_wsbh: str = ""  # 文书编号
    c_fybh: str = ""  # 法院编号
    c_fymc: str = ""  # 法院名称
    dt_cjsj: str = ""  # 创建时间（ISO格式）

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "DocumentDetail":
        """从 API 响应字典创建实例"""
        return cls(
            c_sdbh=data.get("c_sdbh", ""),
            c_wsmc=data.get("c_wsmc", ""),
            c_wjgs=data.get("c_wjgs", ""),
            wjlj=data.get("wjlj", ""),
            c_stbh=data.get("c_stbh", ""),
            c_wsbh=data.get("c_wsbh", ""),
            c_fybh=data.get("c_fybh", ""),
            c_fymc=data.get("c_fymc", ""),
            dt_cjsj=data.get("dt_cjsj", ""),
        )

    def to_dict(self) -> dict[str, Any]:
        """序列化为字典"""
        return {
            "c_sdbh": self.c_sdbh,
            "c_wsmc": self.c_wsmc,
            "c_wjgs": self.c_wjgs,
            "wjlj": self.wjlj,
            "c_stbh": self.c_stbh,
            "c_wsbh": self.c_wsbh,
            "c_fybh": self.c_fybh,
            "c_fymc": self.c_fymc,
            "dt_cjsj": self.dt_cjsj,
        }


@dataclass
class DocumentListResponse:
    """
    文书列表 API 响应
    对应 Requirements 1.4, 3.4
    """

    total: int  # data.total - 总数量，用于分页计算
    documents: list[DocumentRecord] = field(default_factory=list)  # data.data - 文书记录列表

    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "DocumentListResponse":
        """
        从 API 响应字典创建实例
        API 响应格式:
        {
            "code": 200,
            "msg": "成功！",
            "success": true,
            "data": {
                "total": 19,
                "data": [...]
            }
        }
        """
        inner_data = data.get("data", {})
        total = inner_data.get("total", 0)
        documents_data = inner_data.get("data", [])

        documents = [DocumentRecord.from_api_response(doc) for doc in documents_data]

        return cls(total=total, documents=documents)

    def to_dict(self) -> dict[str, Any]:
        """序列化为字典"""
        return {
            "total": self.total,
            "documents": [doc.to_dict() for doc in self.documents],
        }


# ============================================================
# 原有数据类 (Playwright 方式使用)
# ============================================================


@dataclass
class DocumentDeliveryRecord:
    """文书送达记录"""

    case_number: str  # 案号
    send_time: datetime | None  # 发送时间
    element_index: int  # 页面元素索引（用于定位下载按钮）
    document_name: str = ""  # 文书名称（可选）
    court_name: str = ""  # 法院名称（可选）
    delivery_event_id: str = ""  # 送达事件标识（优先使用 sdbh）

    def to_dict(self) -> dict[str, Any]:
        """序列化为字典"""
        return {
            "case_number": self.case_number,
            "send_time": self.send_time.isoformat() if self.send_time else None,
            "element_index": self.element_index,
            "document_name": self.document_name,
            "court_name": self.court_name,
            "delivery_event_id": self.delivery_event_id,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "DocumentDeliveryRecord":
        """从字典反序列化"""
        send_time = None
        if data.get("send_time"):
            if isinstance(data["send_time"], str):
                send_time = datetime.fromisoformat(data["send_time"])
            else:
                send_time = data["send_time"]

        return cls(
            case_number=data["case_number"],
            send_time=send_time,
            element_index=data["element_index"],
            document_name=data.get("document_name", ""),
            court_name=data.get("court_name", ""),
            delivery_event_id=data.get("delivery_event_id", ""),
        )


@dataclass
class DocumentQueryResult:
    """文书查询结果"""

    total_found: int  # 发现的文书总数
    processed_count: int  # 处理的文书数
    skipped_count: int  # 跳过的文书数（时间过滤或已处理）
    failed_count: int  # 处理失败数
    case_log_ids: list[int]  # 创建的案件日志 ID 列表
    errors: list[str]  # 错误信息列表


@dataclass
class DocumentProcessResult:
    """单个文书处理结果"""

    success: bool  # 是否成功
    case_id: int | None  # 匹配的案件ID
    case_log_id: int | None  # 创建的案件日志ID
    renamed_path: str | None  # 重命名后的文件路径
    notification_sent: bool  # 是否发送了通知
    error_message: str | None  # 错误信息

	"""
	文书送达数据类

	定义文书送达相关的数据传输对象
	"""

	from dataclasses import dataclass, field
	from datetime import datetime
	from typing import Any

	# ============================================================
	# API 响应数据类 (来自法院一张网 API)
	# ============================================================


	@dataclass
	class DocumentRecord:
	"""
	单个文书记录 - 来自 getSdListByZjhmAndAhdmNew API

	对应 Requirements 1.2, 3.1
	"""

	ah: str # 案号，如"（2025）粤0604民初41257号"
	sdbh: str # 送达编号 - 用于获取文书详情
	ajzybh: str # 案件主要编号
	fssj: str # 发送时间，如"2025-12-10 16:25:37" - 用于时间过滤
	fymc: str # 法院名称
	ahdm: str = "" # 案号代码
	fybh: str = "" # 法院编号
	ssdrxm: str = "" # 送达人姓名
	ssdrsjhm: str = "" # 送达人手机号
	ssdrzjhm: str = "" # 送达人证件号码
	wsmc: str = "" # 文书名称（多个用逗号分隔）
	sdzt: str = "" # 送达状态
	qdzt: str = "" # 签到状态
	qdbh: str = "" # 签到编号
	fqr: str = "" # 发起人
	cjsj: str = "" # 创建时间
	zhxgsj: str = "" # 最后修改时间

	def parse_fssj(self) -> datetime \| None:
	"""
	解析 fssj（发送时间）字符串为 datetime 对象

	Returns:
	datetime 对象，解析失败返回 None
	"""
	if not self.fssj:
	return None
	try:
	return datetime.strptime(self.fssj, "%Y-%m-%d %H:%M:%S")
	except ValueError:
	# 尝试其他格式
	try:
	return datetime.fromisoformat(self.fssj)
	except ValueError:
	return None

	@classmethod
	def from_api_response(cls, data: dict[str, Any]) -> "DocumentRecord":
	"""从 API 响应字典创建实例"""
	return cls(
	ah=data.get("ah", ""),
	sdbh=data.get("sdbh", ""),
	ajzybh=data.get("ajzybh", ""),
	fssj=data.get("fssj", ""),
	fymc=data.get("fymc", ""),
	ahdm=data.get("ahdm", ""),
	fybh=data.get("fybh", ""),
	ssdrxm=data.get("ssdrxm", ""),
	ssdrsjhm=data.get("ssdrsjhm", ""),
	ssdrzjhm=data.get("ssdrzjhm", ""),
	wsmc=data.get("wsmc", ""),
	sdzt=data.get("sdzt", ""),
	qdzt=data.get("qdzt", ""),
	qdbh=data.get("qdbh", ""),
	fqr=data.get("fqr", ""),
	cjsj=data.get("cjsj", ""),
	zhxgsj=data.get("zhxgsj", ""),
	)

	def to_dict(self) -> dict[str, Any]:
	"""序列化为字典"""
	return {
	"ah": self.ah,
	"sdbh": self.sdbh,
	"ajzybh": self.ajzybh,
	"fssj": self.fssj,
	"fymc": self.fymc,
	"ahdm": self.ahdm,
	"fybh": self.fybh,
	"ssdrxm": self.ssdrxm,
	"ssdrsjhm": self.ssdrsjhm,
	"ssdrzjhm": self.ssdrzjhm,
	"wsmc": self.wsmc,
	"sdzt": self.sdzt,
	"qdzt": self.qdzt,
	"qdbh": self.qdbh,
	"fqr": self.fqr,
	"cjsj": self.cjsj,
	"zhxgsj": self.zhxgsj,
	}


	@dataclass
	class DocumentDetail:
	"""
	文书详情（下载信息）- 来自 getWsListBySdbhNew API

	对应 Requirements 2.2
	"""

	c_sdbh: str # 送达编号
	c_wsmc: str # 文书名称
	c_wjgs: str # 文件格式（如 pdf）
	wjlj: str # 文件链接（OSS URL，带签名）
	c_stbh: str = "" # 上传编号（文件路径）
	c_wsbh: str = "" # 文书编号
	c_fybh: str = "" # 法院编号
	c_fymc: str = "" # 法院名称
	dt_cjsj: str = "" # 创建时间（ISO格式）

	@classmethod
	def from_api_response(cls, data: dict[str, Any]) -> "DocumentDetail":
	"""从 API 响应字典创建实例"""
	return cls(
	c_sdbh=data.get("c_sdbh", ""),
	c_wsmc=data.get("c_wsmc", ""),
	c_wjgs=data.get("c_wjgs", ""),
	wjlj=data.get("wjlj", ""),
	c_stbh=data.get("c_stbh", ""),
	c_wsbh=data.get("c_wsbh", ""),
	c_fybh=data.get("c_fybh", ""),
	c_fymc=data.get("c_fymc", ""),
	dt_cjsj=data.get("dt_cjsj", ""),
	)

	def to_dict(self) -> dict[str, Any]:
	"""序列化为字典"""
	return {
	"c_sdbh": self.c_sdbh,
	"c_wsmc": self.c_wsmc,
	"c_wjgs": self.c_wjgs,
	"wjlj": self.wjlj,
	"c_stbh": self.c_stbh,
	"c_wsbh": self.c_wsbh,
	"c_fybh": self.c_fybh,
	"c_fymc": self.c_fymc,
	"dt_cjsj": self.dt_cjsj,
	}


	@dataclass
	class DocumentListResponse:
	"""
	文书列表 API 响应

	对应 Requirements 1.4, 3.4
	"""

	total: int # data.total - 总数量，用于分页计算
	documents: list[DocumentRecord] = field(default_factory=list) # data.data - 文书记录列表

	@classmethod
	def from_api_response(cls, data: dict[str, Any]) -> "DocumentListResponse":
	"""
	从 API 响应字典创建实例

	API 响应格式:
	{
	"code": 200,
	"msg": "成功！",
	"success": true,
	"data": {
	"total": 19,
	"data": [...]
	}
	}
	"""
	inner_data = data.get("data", {})
	total = inner_data.get("total", 0)
	documents_data = inner_data.get("data", [])

	documents = [DocumentRecord.from_api_response(doc) for doc in documents_data]

	return cls(total=total, documents=documents)

	def to_dict(self) -> dict[str, Any]:
	"""序列化为字典"""
	return {
	"total": self.total,
	"documents": [doc.to_dict() for doc in self.documents],
	}


	# ============================================================
	# 原有数据类 (Playwright 方式使用)
	# ============================================================


	@dataclass
	class DocumentDeliveryRecord:
	"""文书送达记录"""

	case_number: str # 案号
	send_time: datetime \| None # 发送时间
	element_index: int # 页面元素索引（用于定位下载按钮）
	document_name: str = "" # 文书名称（可选）
	court_name: str = "" # 法院名称（可选）
	delivery_event_id: str = "" # 送达事件标识（优先使用 sdbh）

	def to_dict(self) -> dict[str, Any]:
	"""序列化为字典"""
	return {
	"case_number": self.case_number,
	"send_time": self.send_time.isoformat() if self.send_time else None,
	"element_index": self.element_index,
	"document_name": self.document_name,
	"court_name": self.court_name,
	"delivery_event_id": self.delivery_event_id,
	}

	@classmethod
	def from_dict(cls, data: dict[str, Any]) -> "DocumentDeliveryRecord":
	"""从字典反序列化"""
	send_time = None
	if data.get("send_time"):
	if isinstance(data["send_time"], str):
	send_time = datetime.fromisoformat(data["send_time"])
	else:
	send_time = data["send_time"]

	return cls(
	case_number=data["case_number"],
	send_time=send_time,
	element_index=data["element_index"],
	document_name=data.get("document_name", ""),
	court_name=data.get("court_name", ""),
	delivery_event_id=data.get("delivery_event_id", ""),
	)


	@dataclass
	class DocumentQueryResult:
	"""文书查询结果"""

	total_found: int # 发现的文书总数
	processed_count: int # 处理的文书数
	skipped_count: int # 跳过的文书数（时间过滤或已处理）
	failed_count: int # 处理失败数
	case_log_ids: list[int] # 创建的案件日志 ID 列表
	errors: list[str] # 错误信息列表


	@dataclass
	class DocumentProcessResult:
	"""单个文书处理结果"""

	success: bool # 是否成功
	case_id: int \| None # 匹配的案件ID
	case_log_id: int \| None # 创建的案件日志ID
	renamed_path: str \| None # 重命名后的文件路径
	notification_sent: bool # 是否发送了通知
	error_message: str \| None # 错误信息