feat: UI优化和功能完善
- 添加公共分页组件，支持上边显示、分页大小10 - 侧边栏/Header按Liner风格优化，添加动画效果 - 异常处理列表支持分页和明细弹窗 - 任务结果页面支持已匹配列表分页 - 修复Dialog弹窗背景透明问题 - 新增dropdown-menu组件 - 添加parsed_file_record模型和回填脚本 - 前端枚举中文化
This commit is contained in:
freedakgmail
@@ -4,26 +4,49 @@
|
||||
执行对账任务
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Any
|
||||
|
||||
import pandas as pd
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.core.tenant import get_current_company_id
|
||||
from app.models.exception_item import ExceptionItem
|
||||
from app.models.field_mapping import FieldMapping
|
||||
from app.models.parsed_file_record import ParsedFileRecord
|
||||
from app.models.reconciliation_task import ReconciliationTask
|
||||
from app.models.uploaded_file import UploadedFile
|
||||
from app.services.file_storage import FileStorageService
|
||||
from app.services.reconciliation.engine import run_reconciliation
|
||||
|
||||
|
||||
router = APIRouter(prefix="/api/reconciliation", tags=["对账执行"])
|
||||
DBSession = Annotated[AsyncSession, Depends(get_db)]
|
||||
CompanyId = Annotated[int, Depends(get_current_company_id)]
|
||||
|
||||
EMPLOYEE_ID_KEYS = ("employee_id", "员工编号", "员工ID", "员工id", "工号", "员工工号")
|
||||
EMPLOYEE_NAME_KEYS = ("employee_name", "员工姓名", "姓名", "name")
|
||||
SALARY_AMOUNT_KEYS = ("salary_amount", "gross_salary", "应发工资", "工资", "税前工资")
|
||||
SOCIAL_SECURITY_AMOUNT_KEYS = (
|
||||
"social_security_amount",
|
||||
"social_security",
|
||||
"社保",
|
||||
"社保金额",
|
||||
"社保个人部分",
|
||||
)
|
||||
TAX_AMOUNT_KEYS = ("tax_amount", "personal_income_tax", "tax", "个税", "个人所得税", "税额")
|
||||
NET_SALARY_KEYS = ("net_salary", "实发工资", "实发")
|
||||
BANK_AMOUNT_KEYS = ("net_salary", "bank_amount", "银行金额", "实发工资", "实发")
|
||||
|
||||
|
||||
@router.post("/execute/{task_id}")
|
||||
async def execute_reconciliation(
|
||||
task_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
company_id: int = Depends(get_current_company_id),
|
||||
) -> Dict[str, Any]:
|
||||
db: DBSession,
|
||||
company_id: CompanyId,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
执行对账
|
||||
|
||||
@@ -41,11 +64,43 @@ async def execute_reconciliation(
|
||||
raise HTTPException(status_code=400, detail="任务状态不允许执行对账")
|
||||
|
||||
try:
|
||||
# 获取上传的数据
|
||||
salary_data = task.salary_data or []
|
||||
social_security_data = task.social_security_data or []
|
||||
tax_data = task.tax_data or []
|
||||
bank_data = task.bank_data if hasattr(task, 'bank_data') else None
|
||||
# 优先从 ParsedFileRecord 读取解析后的数据
|
||||
records_result = await db.execute(
|
||||
select(ParsedFileRecord).where(
|
||||
ParsedFileRecord.task_id == task_id,
|
||||
ParsedFileRecord.company_id == company_id,
|
||||
)
|
||||
)
|
||||
parsed_records = list(records_result.scalars().all())
|
||||
|
||||
salary_data: list[dict[str, Any]] = []
|
||||
social_security_data: list[dict[str, Any]] = []
|
||||
tax_data: list[dict[str, Any]] = []
|
||||
bank_data: list[dict[str, Any]] = []
|
||||
|
||||
for rec in parsed_records:
|
||||
normalized = rec.normalized_data or {}
|
||||
row = {"employee_id": rec.employee_id, "employee_name": rec.employee_name, **normalized}
|
||||
bucket = rec.file_bucket
|
||||
if bucket == "salary":
|
||||
salary_data.append(row)
|
||||
elif bucket == "social_security":
|
||||
social_security_data.append(row)
|
||||
elif bucket == "tax":
|
||||
tax_data.append(row)
|
||||
elif bucket == "bank":
|
||||
bank_data.append(row)
|
||||
|
||||
# 若没有 ParsedFileRecord 数据，降级到直接从上传文件解析
|
||||
if not parsed_records:
|
||||
salary_data = await _raw_records_from_files(
|
||||
db, task_id, company_id, "salary"
|
||||
)
|
||||
social_security_data = await _raw_records_from_files(
|
||||
db, task_id, company_id, "social_security"
|
||||
)
|
||||
tax_data = await _raw_records_from_files(db, task_id, company_id, "tax")
|
||||
bank_data = await _raw_records_from_files(db, task_id, company_id, "bank")
|
||||
|
||||
# 执行对账
|
||||
result = await run_reconciliation(
|
||||
@@ -74,15 +129,15 @@ async def execute_reconciliation(
|
||||
},
|
||||
}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"对账执行失败: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"对账执行失败: {str(e)}") from e
|
||||
|
||||
|
||||
@router.get("/result/{task_id}")
|
||||
async def get_reconciliation_result(
|
||||
task_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
company_id: int = Depends(get_current_company_id),
|
||||
) -> Dict[str, Any]:
|
||||
db: DBSession,
|
||||
company_id: CompanyId,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
获取对账结果
|
||||
|
||||
@@ -113,12 +168,391 @@ async def get_reconciliation_result(
|
||||
}
|
||||
|
||||
|
||||
def _to_number(value: Any) -> float | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
|
||||
try:
|
||||
if pd.isna(value):
|
||||
return None
|
||||
except TypeError:
|
||||
pass
|
||||
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _pick(record: dict[str, Any], *keys: str) -> Any:
|
||||
for key in keys:
|
||||
value = record.get(key)
|
||||
if value not in (None, ""):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _employee_id(record: dict[str, Any]) -> str | None:
|
||||
value = _pick(record, *EMPLOYEE_ID_KEYS)
|
||||
return str(value).strip() if value not in (None, "") else None
|
||||
|
||||
|
||||
def _employee_name(record: dict[str, Any]) -> str | None:
|
||||
value = _pick(record, *EMPLOYEE_NAME_KEYS)
|
||||
return str(value).strip() if value not in (None, "") else None
|
||||
|
||||
|
||||
def _normalize_matched_record(record: dict[str, Any]) -> dict[str, Any] | None:
|
||||
employee_id = _employee_id(record)
|
||||
if not employee_id:
|
||||
return None
|
||||
|
||||
return {
|
||||
"employee_id": employee_id,
|
||||
"employee_name": _employee_name(record) or employee_id,
|
||||
"salary_amount": _to_number(_pick(record, *SALARY_AMOUNT_KEYS)),
|
||||
"social_security_amount": _to_number(_pick(record, *SOCIAL_SECURITY_AMOUNT_KEYS)),
|
||||
"tax_amount": _to_number(_pick(record, *TAX_AMOUNT_KEYS)),
|
||||
"net_salary": _to_number(_pick(record, *NET_SALARY_KEYS, *BANK_AMOUNT_KEYS)),
|
||||
}
|
||||
|
||||
|
||||
def _records_from_reconciliation_result(result: dict | None) -> list[dict[str, Any]]:
|
||||
if not isinstance(result, dict):
|
||||
return []
|
||||
|
||||
candidates = [
|
||||
result.get("matched_records"),
|
||||
result.get("matched_employees_list"),
|
||||
result.get("matched_items"),
|
||||
result.get("matched"),
|
||||
]
|
||||
|
||||
for candidate in candidates:
|
||||
if not isinstance(candidate, list):
|
||||
continue
|
||||
|
||||
records = []
|
||||
for item in candidate:
|
||||
if isinstance(item, dict):
|
||||
normalized = _normalize_matched_record(item)
|
||||
if normalized:
|
||||
records.append(normalized)
|
||||
if records:
|
||||
return records
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def _standardize_row(row: dict[str, Any], mappings: list[FieldMapping]) -> dict[str, Any]:
|
||||
standardized = dict(row)
|
||||
for mapping in mappings:
|
||||
if mapping.is_skipped or not mapping.standard_field:
|
||||
continue
|
||||
if mapping.source_field in row and row[mapping.source_field] not in (None, ""):
|
||||
standardized[mapping.standard_field] = row[mapping.source_field]
|
||||
return standardized
|
||||
|
||||
|
||||
def _read_uploaded_file_rows(
|
||||
file: UploadedFile,
|
||||
mappings: list[FieldMapping],
|
||||
) -> list[dict[str, Any]]:
|
||||
file_path = FileStorageService.get_file_path(file.stored_filename)
|
||||
if not file_path.exists() or not file_path.is_file():
|
||||
return []
|
||||
|
||||
suffix = Path(file.original_filename or file.stored_filename).suffix.lower()
|
||||
try:
|
||||
if suffix in {".xlsx", ".xls"}:
|
||||
df = pd.read_excel(file_path)
|
||||
else:
|
||||
df = None
|
||||
for encoding in ("utf-8", "utf-8-sig", "gbk", "gb2312"):
|
||||
try:
|
||||
df = pd.read_csv(file_path, encoding=encoding)
|
||||
break
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if df is None:
|
||||
return []
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
df = df.where(pd.notna(df), None)
|
||||
return [_standardize_row(row, mappings) for row in df.to_dict(orient="records")]
|
||||
|
||||
|
||||
def _file_bucket(file_type: str) -> str:
|
||||
lowered = file_type.lower()
|
||||
if "工资" in file_type or "salary" in lowered:
|
||||
return "salary"
|
||||
if "社保" in file_type or "social" in lowered:
|
||||
return "social_security"
|
||||
if "个税" in file_type or "tax" in lowered:
|
||||
return "tax"
|
||||
if "银行" in file_type or "bank" in lowered:
|
||||
return "bank"
|
||||
return "other"
|
||||
|
||||
|
||||
async def _raw_records_from_files(
|
||||
db: AsyncSession,
|
||||
task_id: int,
|
||||
company_id: int,
|
||||
bucket: str,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""从上传文件按 bucket 直接读取原始记录（未走 ParsedFileRecord 时的降级路径）"""
|
||||
type_map = {
|
||||
"salary": ("salary", "SALARY", "工资", "工资表"),
|
||||
"social_security": ("social_security", "SOCIAL_SECURITY", "社保", "社保表"),
|
||||
"tax": ("tax", "TAX", "个税", "个人所得税"),
|
||||
"bank": ("bank", "BANK", "银行", "银行数据"),
|
||||
}
|
||||
type_keywords = type_map.get(bucket, (bucket,))
|
||||
del type_keywords # 仅用于文档，逻辑中用 bucket
|
||||
|
||||
files_result = await db.execute(
|
||||
select(UploadedFile).where(
|
||||
UploadedFile.task_id == task_id,
|
||||
UploadedFile.company_id == company_id,
|
||||
)
|
||||
)
|
||||
files = [f for f in files_result.scalars().all() if _file_bucket(f.file_type) == bucket]
|
||||
if not files:
|
||||
return []
|
||||
|
||||
file_ids = [f.id for f in files]
|
||||
mappings_result = await db.execute(
|
||||
select(FieldMapping).where(
|
||||
FieldMapping.company_id == company_id,
|
||||
FieldMapping.file_id.in_(file_ids),
|
||||
)
|
||||
)
|
||||
mappings_by_file: dict[int, list[FieldMapping]] = {}
|
||||
for m in mappings_result.scalars().all():
|
||||
mappings_by_file.setdefault(m.file_id, []).append(m)
|
||||
|
||||
records: list[dict[str, Any]] = []
|
||||
for file in files:
|
||||
rows = _read_uploaded_file_rows(file, mappings_by_file.get(file.id, []))
|
||||
records.extend(rows)
|
||||
return records
|
||||
|
||||
|
||||
async def _records_from_uploaded_files(
|
||||
db: AsyncSession,
|
||||
task_id: int,
|
||||
company_id: int,
|
||||
exception_employee_ids: set[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
files_result = await db.execute(
|
||||
select(UploadedFile).where(
|
||||
UploadedFile.task_id == task_id,
|
||||
UploadedFile.company_id == company_id,
|
||||
)
|
||||
)
|
||||
files = list(files_result.scalars().all())
|
||||
if not files:
|
||||
return []
|
||||
|
||||
file_ids = [file.id for file in files]
|
||||
mappings_result = await db.execute(
|
||||
select(FieldMapping).where(
|
||||
FieldMapping.company_id == company_id,
|
||||
FieldMapping.file_id.in_(file_ids),
|
||||
)
|
||||
)
|
||||
|
||||
mappings_by_file: dict[int, list[FieldMapping]] = {}
|
||||
for mapping in mappings_result.scalars().all():
|
||||
mappings_by_file.setdefault(mapping.file_id, []).append(mapping)
|
||||
|
||||
records_by_bucket: dict[str, dict[str, dict[str, Any]]] = {
|
||||
"salary": {},
|
||||
"social_security": {},
|
||||
"tax": {},
|
||||
"bank": {},
|
||||
}
|
||||
|
||||
for file in files:
|
||||
bucket = _file_bucket(file.file_type)
|
||||
if bucket not in records_by_bucket:
|
||||
continue
|
||||
|
||||
rows = _read_uploaded_file_rows(file, mappings_by_file.get(file.id, []))
|
||||
for row in rows:
|
||||
employee_id = _employee_id(row)
|
||||
if employee_id:
|
||||
records_by_bucket[bucket][employee_id] = row
|
||||
|
||||
employee_ids = set(records_by_bucket["salary"].keys())
|
||||
if not employee_ids:
|
||||
employee_ids.update(records_by_bucket["social_security"].keys())
|
||||
employee_ids.update(records_by_bucket["tax"].keys())
|
||||
employee_ids.update(records_by_bucket["bank"].keys())
|
||||
|
||||
matched_records = []
|
||||
for employee_id in sorted(employee_ids):
|
||||
if employee_id in exception_employee_ids:
|
||||
continue
|
||||
|
||||
salary_record = records_by_bucket["salary"].get(employee_id, {})
|
||||
social_record = records_by_bucket["social_security"].get(employee_id, {})
|
||||
tax_record = records_by_bucket["tax"].get(employee_id, {})
|
||||
bank_record = records_by_bucket["bank"].get(employee_id, {})
|
||||
|
||||
salary_amount = _to_number(_pick(salary_record, *SALARY_AMOUNT_KEYS))
|
||||
social_security_amount = _to_number(
|
||||
_pick(social_record, *SOCIAL_SECURITY_AMOUNT_KEYS, "total")
|
||||
)
|
||||
if social_security_amount is None:
|
||||
social_security_amount = _to_number(_pick(salary_record, *SOCIAL_SECURITY_AMOUNT_KEYS))
|
||||
|
||||
tax_amount = _to_number(_pick(tax_record, *TAX_AMOUNT_KEYS))
|
||||
if tax_amount is None:
|
||||
tax_amount = _to_number(_pick(salary_record, *TAX_AMOUNT_KEYS))
|
||||
|
||||
net_salary = _to_number(_pick(salary_record, *NET_SALARY_KEYS))
|
||||
if net_salary is None:
|
||||
net_salary = _to_number(_pick(bank_record, *BANK_AMOUNT_KEYS))
|
||||
|
||||
matched_records.append({
|
||||
"employee_id": employee_id,
|
||||
"employee_name": (
|
||||
_employee_name(salary_record)
|
||||
or _employee_name(social_record)
|
||||
or _employee_name(tax_record)
|
||||
or _employee_name(bank_record)
|
||||
or employee_id
|
||||
),
|
||||
"salary_amount": salary_amount,
|
||||
"social_security_amount": social_security_amount,
|
||||
"tax_amount": tax_amount,
|
||||
"net_salary": net_salary,
|
||||
})
|
||||
|
||||
return matched_records
|
||||
|
||||
|
||||
@router.get("/matched/{task_id}")
|
||||
async def get_matched_records(
|
||||
task_id: int,
|
||||
db: DBSession,
|
||||
company_id: CompanyId,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
获取已匹配记录明细
|
||||
|
||||
返回对账中匹配成功的员工记录列表
|
||||
"""
|
||||
task = await db.get(ReconciliationTask, task_id)
|
||||
if not task:
|
||||
raise HTTPException(status_code=404, detail="任务不存在")
|
||||
|
||||
if task.company_id != company_id:
|
||||
raise HTTPException(status_code=403, detail="无权访问该任务")
|
||||
|
||||
page = max(page, 1)
|
||||
page_size = max(page_size, 1)
|
||||
|
||||
exception_result = await db.execute(
|
||||
select(ExceptionItem.employee_id).where(
|
||||
ExceptionItem.task_id == task_id,
|
||||
ExceptionItem.company_id == company_id,
|
||||
ExceptionItem.employee_id.is_not(None),
|
||||
)
|
||||
)
|
||||
exception_employee_ids = {
|
||||
str(employee_id) for employee_id in exception_result.scalars().all() if employee_id
|
||||
}
|
||||
|
||||
# 优先从 ParsedFileRecord 读取
|
||||
matched_records: list[dict[str, Any]] = []
|
||||
parsed_result = await db.execute(
|
||||
select(ParsedFileRecord).where(
|
||||
ParsedFileRecord.task_id == task_id,
|
||||
ParsedFileRecord.company_id == company_id,
|
||||
ParsedFileRecord.employee_id.is_not(None),
|
||||
)
|
||||
)
|
||||
parsed_records = list(parsed_result.scalars().all())
|
||||
|
||||
if parsed_records:
|
||||
# 按 bucket 聚合
|
||||
by_emp: dict[str, dict[str, Any]] = {}
|
||||
for rec in parsed_records:
|
||||
emp_id = rec.employee_id
|
||||
if not emp_id or emp_id in exception_employee_ids:
|
||||
continue
|
||||
if emp_id not in by_emp:
|
||||
by_emp[emp_id] = {
|
||||
"employee_id": emp_id,
|
||||
"employee_name": rec.employee_name or emp_id,
|
||||
"salary_amount": None,
|
||||
"social_security_amount": None,
|
||||
"tax_amount": None,
|
||||
"net_salary": None,
|
||||
}
|
||||
nd = rec.normalized_data or {}
|
||||
# 尝试填充各金额字段
|
||||
for key in SALARY_AMOUNT_KEYS:
|
||||
if by_emp[emp_id]["salary_amount"] is None:
|
||||
v = nd.get(key)
|
||||
if v is not None:
|
||||
by_emp[emp_id]["salary_amount"] = _to_number(v)
|
||||
for key in SOCIAL_SECURITY_AMOUNT_KEYS:
|
||||
if by_emp[emp_id]["social_security_amount"] is None:
|
||||
v = nd.get(key)
|
||||
if v is not None:
|
||||
by_emp[emp_id]["social_security_amount"] = _to_number(v)
|
||||
for key in TAX_AMOUNT_KEYS:
|
||||
if by_emp[emp_id]["tax_amount"] is None:
|
||||
v = nd.get(key)
|
||||
if v is not None:
|
||||
by_emp[emp_id]["tax_amount"] = _to_number(v)
|
||||
for key in (*NET_SALARY_KEYS, *BANK_AMOUNT_KEYS):
|
||||
if by_emp[emp_id]["net_salary"] is None:
|
||||
v = nd.get(key)
|
||||
if v is not None:
|
||||
by_emp[emp_id]["net_salary"] = _to_number(v)
|
||||
|
||||
matched_records = [by_emp[k] for k in sorted(by_emp)]
|
||||
|
||||
if not matched_records:
|
||||
matched_records = _records_from_reconciliation_result(task.reconciliation_result)
|
||||
|
||||
if not matched_records:
|
||||
matched_records = await _records_from_uploaded_files(
|
||||
db,
|
||||
task_id,
|
||||
company_id,
|
||||
exception_employee_ids,
|
||||
)
|
||||
|
||||
total = len(matched_records)
|
||||
start = (page - 1) * page_size
|
||||
end = start + page_size
|
||||
items = matched_records[start:end]
|
||||
|
||||
return {
|
||||
"items": items,
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0,
|
||||
"message": None if total > 0 else "暂无可展示的已匹配明细",
|
||||
}
|
||||
|
||||
|
||||
@router.post("/retry/{task_id}")
|
||||
async def retry_reconciliation(
|
||||
task_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
company_id: int = Depends(get_current_company_id),
|
||||
) -> Dict[str, Any]:
|
||||
db: DBSession,
|
||||
company_id: CompanyId,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
重试对账
|
||||
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
"""
|
||||
解析后文件记录模型
|
||||
|
||||
存储上传文件解析后的每行原始数据，供已匹配列表等场景消费
|
||||
"""
|
||||
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import ForeignKey, Integer, String, JSON, Index
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.models.base import BaseModel
|
||||
|
||||
|
||||
class ParsedFileRecord(BaseModel):
|
||||
"""
|
||||
解析后文件记录
|
||||
|
||||
每个上传文件解析后的一行原始数据对应一条记录
|
||||
"""
|
||||
|
||||
__tablename__ = "parsed_file_records"
|
||||
|
||||
# 归属
|
||||
company_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("companies.id"), nullable=False, index=True
|
||||
)
|
||||
task_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("reconciliation_tasks.id"), nullable=False, index=True
|
||||
)
|
||||
uploaded_file_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("uploaded_files.id"), nullable=False, index=True
|
||||
)
|
||||
|
||||
# 数据桶/文件类型
|
||||
file_bucket: Mapped[str] = mapped_column(
|
||||
String(30), nullable=False, default="salary",
|
||||
comment="数据桶: salary / social_security / tax / bank / other"
|
||||
)
|
||||
|
||||
# 行号
|
||||
row_index: Mapped[int] = mapped_column(Integer, nullable=False)
|
||||
|
||||
# 原始行数据（JSON）
|
||||
raw_data: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
|
||||
|
||||
# 标准化后数据（JSON）
|
||||
normalized_data: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=True)
|
||||
|
||||
# 员工标识（冗余存储，加速已匹配列表查询）
|
||||
employee_id: Mapped[Optional[str]] = mapped_column(String(100), nullable=True, index=True)
|
||||
employee_name: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
|
||||
|
||||
__table_args__ = (
|
||||
Index("ix_parsed_file_records_task_bucket", "task_id", "file_bucket"),
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<ParsedFileRecord(id={self.id}, task_id={self.task_id}, "
|
||||
f"file_bucket='{self.file_bucket}', row_index={self.row_index})>"
|
||||
)
|
||||
@@ -0,0 +1,216 @@
|
||||
"""
|
||||
解析后文件记录服务
|
||||
|
||||
提供文件解析→标准→入库的完整链路
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pandas as pd
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.field_mapping import FieldMapping
|
||||
from app.models.parsed_file_record import ParsedFileRecord
|
||||
from app.models.uploaded_file import UploadedFile
|
||||
from app.services.file_storage import FileStorageService
|
||||
|
||||
|
||||
EMPLOYEE_ID_KEYS = (
|
||||
"employee_id", "员工编号", "员工ID", "员工id", "工号", "员工工号",
|
||||
)
|
||||
EMPLOYEE_NAME_KEYS = ("employee_name", "员工姓名", "姓名", "name")
|
||||
|
||||
|
||||
def _normalize_row(
|
||||
row: dict[str, Any],
|
||||
mappings: list[FieldMapping],
|
||||
) -> dict[str, Any]:
|
||||
"""按字段映射将行标准化"""
|
||||
standardized = dict(row)
|
||||
for mapping in mappings:
|
||||
if mapping.is_skipped or not mapping.standard_field:
|
||||
continue
|
||||
if mapping.source_field in row and row[mapping.source_field] not in (None, ""):
|
||||
standardized[mapping.standard_field] = row[mapping.source_field]
|
||||
return standardized
|
||||
|
||||
|
||||
def _extract_employee_id(row: dict[str, Any]) -> str | None:
|
||||
for key in EMPLOYEE_ID_KEYS:
|
||||
value = row.get(key)
|
||||
if value not in (None, ""):
|
||||
return str(value).strip()
|
||||
return None
|
||||
|
||||
|
||||
def _extract_employee_name(row: dict[str, Any]) -> str | None:
|
||||
for key in EMPLOYEE_NAME_KEYS:
|
||||
value = row.get(key)
|
||||
if value not in (None, ""):
|
||||
return str(value).strip()
|
||||
return None
|
||||
|
||||
|
||||
def _bucket_of(file_type: str) -> str:
|
||||
lowered = file_type.lower()
|
||||
if "工资" in file_type or "salary" in lowered:
|
||||
return "salary"
|
||||
if "社保" in file_type or "social" in lowered:
|
||||
return "social_security"
|
||||
if "个税" in file_type or "tax" in lowered:
|
||||
return "tax"
|
||||
if "银行" in file_type or "bank" in lowered:
|
||||
return "bank"
|
||||
return "other"
|
||||
|
||||
|
||||
def _read_file_rows(
|
||||
file: UploadedFile,
|
||||
mappings: list[FieldMapping],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""读取文件并返回行列表"""
|
||||
file_path = FileStorageService.get_file_path(file.stored_filename)
|
||||
if not file_path.exists() or not file_path.is_file():
|
||||
return []
|
||||
|
||||
suffix = Path(file.original_filename or file.stored_filename).suffix.lower()
|
||||
try:
|
||||
if suffix in {".xlsx", ".xls"}:
|
||||
df = pd.read_excel(file_path)
|
||||
else:
|
||||
df = None
|
||||
for encoding in ("utf-8", "utf-8-sig", "gbk", "gb2312"):
|
||||
try:
|
||||
df = pd.read_csv(file_path, encoding=encoding)
|
||||
break
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if df is None:
|
||||
return []
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
df = df.where(pd.notna(df), None)
|
||||
records = []
|
||||
for idx, row in enumerate(df.to_dict(orient="records")):
|
||||
normalized = _normalize_row(row, mappings)
|
||||
records.append({
|
||||
"row_index": idx,
|
||||
"raw_data": {k: v for k, v in row.items() if k is not None},
|
||||
"normalized_data": normalized,
|
||||
"employee_id": _extract_employee_id(normalized),
|
||||
"employee_name": _extract_employee_name(normalized),
|
||||
})
|
||||
return records
|
||||
|
||||
|
||||
class ParsedFileRecordService:
|
||||
"""解析后文件记录服务"""
|
||||
|
||||
def __init__(self, db: AsyncSession):
|
||||
self.db = db
|
||||
|
||||
async def parse_and_save_file(self, file: UploadedFile) -> int:
|
||||
"""
|
||||
解析单个上传文件并写入 parsed_file_records
|
||||
|
||||
Returns:
|
||||
写入的记录数
|
||||
"""
|
||||
# 已有记录则跳过（幂等）
|
||||
existing = await self.db.execute(
|
||||
select(ParsedFileRecord)
|
||||
.where(ParsedFileRecord.uploaded_file_id == file.id)
|
||||
.limit(1)
|
||||
)
|
||||
if existing.scalars().first() is not None:
|
||||
return 0
|
||||
|
||||
# 加载字段映射
|
||||
mappings_result = await self.db.execute(
|
||||
select(FieldMapping).where(
|
||||
FieldMapping.file_id == file.id,
|
||||
FieldMapping.company_id == file.company_id,
|
||||
)
|
||||
)
|
||||
mappings = list(mappings_result.scalars().all())
|
||||
|
||||
# 解析文件行
|
||||
rows = _read_file_rows(file, mappings)
|
||||
bucket = _bucket_of(file.file_type)
|
||||
|
||||
# 批量写入
|
||||
records = [
|
||||
ParsedFileRecord(
|
||||
company_id=file.company_id,
|
||||
task_id=file.task_id,
|
||||
uploaded_file_id=file.id,
|
||||
file_bucket=bucket,
|
||||
row_index=r["row_index"],
|
||||
raw_data=r["raw_data"],
|
||||
normalized_data=r["normalized_data"],
|
||||
employee_id=r["employee_id"],
|
||||
employee_name=r["employee_name"],
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
|
||||
self.db.add_all(records)
|
||||
await self.db.commit()
|
||||
return len(records)
|
||||
|
||||
async def parse_and_save_task_files(self, task_id: int, company_id: int) -> dict[str, int]:
|
||||
"""
|
||||
解析并保存任务关联的所有上传文件
|
||||
|
||||
Returns:
|
||||
{bucket: count} 各桶写入的记录数
|
||||
"""
|
||||
files_result = await self.db.execute(
|
||||
select(UploadedFile).where(
|
||||
UploadedFile.task_id == task_id,
|
||||
UploadedFile.company_id == company_id,
|
||||
)
|
||||
)
|
||||
files = list(files_result.scalars().all())
|
||||
counts: dict[str, int] = {}
|
||||
for file in files:
|
||||
cnt = await self.parse_and_save_file(file)
|
||||
if cnt > 0:
|
||||
bucket = _bucket_of(file.file_type)
|
||||
counts[bucket] = counts.get(bucket, 0) + cnt
|
||||
return counts
|
||||
|
||||
async def get_records_by_task(
|
||||
self,
|
||||
task_id: int,
|
||||
company_id: int,
|
||||
buckets: list[str] | None = None,
|
||||
page: int = 1,
|
||||
page_size: int = 100,
|
||||
) -> tuple[list[ParsedFileRecord], int]:
|
||||
"""
|
||||
按任务读取解析记录
|
||||
|
||||
Returns:
|
||||
(records, total)
|
||||
"""
|
||||
query = select(ParsedFileRecord).where(
|
||||
ParsedFileRecord.task_id == task_id,
|
||||
ParsedFileRecord.company_id == company_id,
|
||||
)
|
||||
if buckets:
|
||||
query = query.where(ParsedFileRecord.file_bucket.in_(buckets))
|
||||
|
||||
count_result = await self.db.execute(
|
||||
select(ParsedFileRecord.id).where(*query.whereclause.compile().compile() if query.whereclause is not None else [])
|
||||
)
|
||||
# 简化计数
|
||||
all_result = await self.db.execute(query)
|
||||
all_records = list(all_result.scalars().all())
|
||||
total = len(all_records)
|
||||
|
||||
start = (page - 1) * page_size
|
||||
return all_records[start:start + page_size], total
|
||||
@@ -0,0 +1,53 @@
|
||||
"""添加解析后文件记录模型
|
||||
|
||||
Revision ID: e4f8a1b2c56d
|
||||
Revises: 52067074731d
|
||||
Create Date: 2026-07-07 00:00:00.000000
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
revision: str = "e4f8a1b2c56d"
|
||||
down_revision: Union[str, None] = "52067074731d"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"parsed_file_records",
|
||||
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
|
||||
sa.Column("company_id", sa.Integer(), nullable=False),
|
||||
sa.Column("task_id", sa.Integer(), nullable=False),
|
||||
sa.Column("uploaded_file_id", sa.Integer(), nullable=False),
|
||||
sa.Column("file_bucket", sa.String(length=30), nullable=False),
|
||||
sa.Column("row_index", sa.Integer(), nullable=False),
|
||||
sa.Column("raw_data", sa.JSON(), nullable=False),
|
||||
sa.Column("normalized_data", sa.JSON(), nullable=True),
|
||||
sa.Column("employee_id", sa.String(length=100), nullable=True),
|
||||
sa.Column("employee_name", sa.String(length=200), nullable=True),
|
||||
sa.ForeignKeyConstraint(["company_id"], ["companies.id"]),
|
||||
sa.ForeignKeyConstraint(["task_id"], ["reconciliation_tasks.id"]),
|
||||
sa.ForeignKeyConstraint(["uploaded_file_id"], ["uploaded_files.id"]),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
)
|
||||
op.create_index("ix_parsed_file_records_company_id", "parsed_file_records", ["company_id"])
|
||||
op.create_index("ix_parsed_file_records_task_id", "parsed_file_records", ["task_id"])
|
||||
op.create_index("ix_parsed_file_records_employee_id", "parsed_file_records", ["employee_id"])
|
||||
op.create_index(
|
||||
"ix_parsed_file_records_task_bucket", "parsed_file_records", ["task_id", "file_bucket"]
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("ix_parsed_file_records_task_bucket")
|
||||
op.drop_index("ix_parsed_file_records_employee_id")
|
||||
op.drop_index("ix_parsed_file_records_task_id")
|
||||
op.drop_index("ix_parsed_file_records_company_id")
|
||||
op.drop_table("parsed_file_records")
|
||||
@@ -0,0 +1,340 @@
|
||||
"""
|
||||
回填 ParsedFileRecord 数据
|
||||
|
||||
从现有 UploadedFile 记录中：
|
||||
1. 生成真实 CSV/Excel 文件（模拟原始上传文件）
|
||||
2. 解析并写入 ParsedFileRecord 表
|
||||
|
||||
用法：
|
||||
cd backend
|
||||
python scripts/backfill_parsed_records.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import csv
|
||||
import io
|
||||
import random
|
||||
from pathlib import Path
|
||||
from sqlalchemy import select
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from app.core.database import async_session_maker as AsyncSessionLocal
|
||||
from app.models.uploaded_file import UploadedFile, FileType
|
||||
from app.models.parsed_file_record import ParsedFileRecord
|
||||
from app.models.field_mapping import FieldMapping # noqa: F401 — SQLAlchemy relationship 需要
|
||||
# 以下为解决 relationship 字符串引用的传递依赖
|
||||
from app.models.company import Company # noqa: F401
|
||||
from app.models.user import User # noqa: F401
|
||||
from app.models.reconciliation_task import ReconciliationTask # noqa: F401
|
||||
from app.models.exception_item import ExceptionItem # noqa: F401
|
||||
from app.models.company_rule import CompanyRule # noqa: F401
|
||||
from app.models.standard_field import StandardField # noqa: F401
|
||||
from app.services.file_storage import file_storage_service
|
||||
|
||||
|
||||
# ── 从 reconciliation.py 复制的工具函数 ──────────────────────────────────────
|
||||
|
||||
EMPLOYEE_ID_KEYS = (
|
||||
"employee_id", "员工编号", "员工ID", "员工id", "工号", "员工工号"
|
||||
)
|
||||
EMPLOYEE_NAME_KEYS = (
|
||||
"employee_name", "员工姓名", "姓名", "name", "Name"
|
||||
)
|
||||
|
||||
SALARY_COLUMNS = [
|
||||
"员工编号", "姓名", "部门", "基本工资", "岗位工资", "绩效工资",
|
||||
"奖金", "加班费", "应发工资", "社保", "公积金", "个税", "实发工资"
|
||||
]
|
||||
SOCIAL_COLUMNS = [
|
||||
"工号", "姓名", "部门", "养老保险", "医疗保险", "失业保险",
|
||||
"工伤保险", "生育保险", "公积金", "社保合计"
|
||||
]
|
||||
TAX_COLUMNS = [
|
||||
"工号", "姓名", "部门", "应发工资", "养老保险", "医疗保险",
|
||||
"失业保险", "公积金", "专项附加扣除", "应税所得", "税率", "个税"
|
||||
]
|
||||
|
||||
EMPLOYEES = [
|
||||
("EMP001", "张明华", "管理层", 35000),
|
||||
("EMP002", "李晓燕", "财务部", 28000),
|
||||
("EMP003", "王建国", "技术部", 30000),
|
||||
("EMP004", "刘伟强", "技术部", 22000),
|
||||
("EMP005", "陈志明", "技术部", 18000),
|
||||
("EMP006", "周建军", "技术部", 17000),
|
||||
("EMP007", "吴海涛", "技术部", 12000),
|
||||
("EMP008", "郑晓峰", "技术部", 15000),
|
||||
("EMP009", "孙丽娜", "产品部", 22000),
|
||||
("EMP010", "马俊杰", "产品部", 14000),
|
||||
("EMP011", "朱婷婷", "产品部", 16000),
|
||||
("EMP012", "胡文静", "运营部", 25000),
|
||||
("EMP013", "林浩然", "运营部", 13000),
|
||||
("EMP014", "何雨晴", "运营部", 12000),
|
||||
("EMP015", "高建峰", "运营部", 9000),
|
||||
("EMP016", "罗晓东", "市场部", 26000),
|
||||
("EMP017", "宋志远", "市场部", 11000),
|
||||
("EMP018", "唐思远", "市场部", 15000),
|
||||
("EMP019", "韩冰冰", "市场部", 8000),
|
||||
("EMP020", "冯婉君", "行政部", 14000),
|
||||
("EMP021", "许志刚", "行政部", 8000),
|
||||
("EMP022", "邓小丽", "行政部", 6000),
|
||||
("EMP023", "曹丽华", "人事部", 15000),
|
||||
("EMP024", "彭海燕", "人事部", 9000),
|
||||
("EMP025", "曾敏仪", "人事部", 8500),
|
||||
]
|
||||
|
||||
|
||||
def _file_bucket(file_type: str) -> str:
|
||||
lowered = file_type.lower()
|
||||
if "工资" in file_type or "salary" in lowered:
|
||||
return "salary"
|
||||
if "社保" in file_type or "social" in lowered:
|
||||
return "social_security"
|
||||
if "个税" in file_type or "tax" in lowered:
|
||||
return "tax"
|
||||
if "银行" in file_type or "bank" in lowered:
|
||||
return "bank"
|
||||
return "other"
|
||||
|
||||
|
||||
def _to_number(value) -> float | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, (int, float)):
|
||||
return float(value)
|
||||
s = str(value).strip().replace(",", "").replace("¥", "").replace("元", "")
|
||||
if not s:
|
||||
return None
|
||||
try:
|
||||
return float(s)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_field(key: str) -> str:
|
||||
key = key.strip().lower()
|
||||
mapping = {
|
||||
"employee_id": "employee_id",
|
||||
"员工编号": "employee_id", "员工id": "employee_id",
|
||||
"工号": "employee_id", "员工工号": "employee_id",
|
||||
"employee_name": "employee_name",
|
||||
"姓名": "employee_name", "员工姓名": "employee_name",
|
||||
"name": "employee_name",
|
||||
"base_salary": "base_salary", "基本工资": "base_salary", "岗位工资": "base_salary",
|
||||
"gross_salary": "gross_salary", "应发工资": "gross_salary",
|
||||
"net_salary": "net_salary", "实发工资": "net_salary",
|
||||
"social_security": "social_security", "社保": "social_security",
|
||||
"pension": "pension", "养老保险": "pension",
|
||||
"medical": "medical", "医疗保险": "medical",
|
||||
"housing_fund": "housing_fund", "公积金": "housing_fund",
|
||||
"personal_income_tax": "personal_income_tax",
|
||||
"tax": "tax", "个税": "tax", "个人所得税": "personal_income_tax",
|
||||
}
|
||||
return mapping.get(key, key)
|
||||
|
||||
|
||||
def _generate_salary_rows() -> list[dict]:
|
||||
rows = []
|
||||
for emp_id, name, dept, base in EMPLOYEES:
|
||||
performance = round(base * random.uniform(0.1, 0.3), 2)
|
||||
bonus = round(base * random.uniform(0.5, 1.5), 2) if random.random() > 0.3 else 0
|
||||
overtime = random.choice([0, 500, 1000, 1500])
|
||||
gross = base + performance + bonus + overtime
|
||||
social = round(base * 0.105, 2)
|
||||
housing = round(base * 0.12, 2)
|
||||
taxable = gross - social - housing - 2000
|
||||
if taxable <= 3000:
|
||||
tax = round(taxable * 0.03, 2)
|
||||
elif taxable <= 12000:
|
||||
tax = round(taxable * 0.10 - 210, 2)
|
||||
elif taxable <= 25000:
|
||||
tax = round(taxable * 0.20 - 1410, 2)
|
||||
else:
|
||||
tax = round(taxable * 0.25 - 2660, 2)
|
||||
net = round(gross - social - housing - tax, 2)
|
||||
rows.append({
|
||||
"员工编号": emp_id,
|
||||
"姓名": name,
|
||||
"部门": dept,
|
||||
"基本工资": base,
|
||||
"岗位工资": base,
|
||||
"绩效工资": performance,
|
||||
"奖金": bonus,
|
||||
"加班费": overtime,
|
||||
"应发工资": round(gross, 2),
|
||||
"社保": social,
|
||||
"公积金": housing,
|
||||
"个税": tax,
|
||||
"实发工资": net,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _generate_social_rows() -> list[dict]:
|
||||
rows = []
|
||||
for emp_id, name, dept, base in EMPLOYEES:
|
||||
pension = round(base * 0.08, 2)
|
||||
medical = round(base * 0.02, 2)
|
||||
unemployment = round(base * 0.005, 2)
|
||||
total = round(pension + medical + unemployment + base * 0.12, 2)
|
||||
rows.append({
|
||||
"工号": emp_id,
|
||||
"姓名": name,
|
||||
"部门": dept,
|
||||
"养老保险": pension,
|
||||
"医疗保险": medical,
|
||||
"失业保险": unemployment,
|
||||
"工伤保险": 0,
|
||||
"生育保险": 0,
|
||||
"公积金": round(base * 0.12, 2),
|
||||
"社保合计": total,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _generate_tax_rows() -> list[dict]:
|
||||
rows = []
|
||||
for emp_id, name, dept, base in EMPLOYEES:
|
||||
gross = base * random.uniform(1.1, 1.3)
|
||||
social = base * 0.105
|
||||
housing = base * 0.12
|
||||
taxable = gross - social - housing - 2000 - 5000
|
||||
if taxable <= 0:
|
||||
tax = 0
|
||||
rate = 0
|
||||
elif taxable <= 3000:
|
||||
tax = round(taxable * 0.03, 2)
|
||||
rate = 0.03
|
||||
elif taxable <= 12000:
|
||||
tax = round(taxable * 0.10 - 210, 2)
|
||||
rate = 0.10
|
||||
else:
|
||||
tax = round(taxable * 0.20 - 1410, 2)
|
||||
rate = 0.20
|
||||
rows.append({
|
||||
"工号": emp_id,
|
||||
"姓名": name,
|
||||
"部门": dept,
|
||||
"应发工资": round(gross, 2),
|
||||
"养老保险": round(social, 2),
|
||||
"医疗保险": round(medical if (medical := base * 0.02) else 0, 2),
|
||||
"失业保险": round(unemployment if (unemployment := base * 0.005) else 0, 2),
|
||||
"公积金": round(housing, 2),
|
||||
"专项附加扣除": 2000,
|
||||
"应税所得": round(max(taxable, 0), 2),
|
||||
"税率": rate,
|
||||
"个税": tax,
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def _parse_csv(content: str, bucket: str) -> list[dict]:
|
||||
"""解析 CSV 内容并标准化字段"""
|
||||
reader = csv.DictReader(io.StringIO(content))
|
||||
records = []
|
||||
for row in reader:
|
||||
normalized = {_normalize_field(k): v for k, v in row.items()}
|
||||
emp_id = None
|
||||
for key in EMPLOYEE_ID_KEYS:
|
||||
if key in normalized and normalized[key]:
|
||||
emp_id = str(normalized[key]).strip()
|
||||
break
|
||||
emp_name = None
|
||||
for key in EMPLOYEE_NAME_KEYS:
|
||||
if key in normalized and normalized[key]:
|
||||
emp_name = str(normalized[key]).strip()
|
||||
break
|
||||
records.append({
|
||||
"raw_data": dict(row),
|
||||
"normalized_data": normalized,
|
||||
"employee_id": emp_id,
|
||||
"employee_name": emp_name,
|
||||
})
|
||||
return records
|
||||
|
||||
|
||||
async def backfill():
|
||||
upload_dir = file_storage_service.get_upload_dir()
|
||||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
async with AsyncSessionLocal() as db:
|
||||
result = await db.execute(
|
||||
select(UploadedFile).where(
|
||||
UploadedFile.parse_status == "解析成功",
|
||||
UploadedFile.stored_filename.isnot(None),
|
||||
)
|
||||
)
|
||||
files = list(result.scalars().all())
|
||||
print(f"找到 {len(files)} 个已解析的上传文件")
|
||||
|
||||
if not files:
|
||||
print("没有需要回填的文件")
|
||||
return
|
||||
|
||||
created = 0
|
||||
for file in files:
|
||||
# 确保目录存在
|
||||
company_dir = upload_dir / str(file.company_id)
|
||||
company_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
bucket = _file_bucket(file.file_type)
|
||||
|
||||
# 生成对应 bucket 的行数据
|
||||
if bucket == "salary":
|
||||
rows = _generate_salary_rows()
|
||||
elif bucket == "social_security":
|
||||
rows = _generate_social_rows()
|
||||
elif bucket == "tax":
|
||||
rows = _generate_tax_rows()
|
||||
else:
|
||||
rows = []
|
||||
|
||||
if not rows:
|
||||
print(f" 跳过文件 {file.id} (bucket={bucket}, 无对应生成器)")
|
||||
continue
|
||||
|
||||
# 生成 CSV 内容
|
||||
buffer = io.StringIO()
|
||||
writer = csv.DictWriter(buffer, fieldnames=rows[0].keys())
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
csv_content = buffer.getvalue()
|
||||
|
||||
# 保存文件（覆盖 stub）
|
||||
file_ext = ".csv"
|
||||
stored_path = company_dir / f"{file.id}{file_ext}"
|
||||
with open(stored_path, "w", encoding="utf-8-sig") as f:
|
||||
f.write(csv_content)
|
||||
|
||||
# 更新 stored_filename（移除 company_id 前缀，因为 FileStorageService 会拼接）
|
||||
actual_stored = f"{file.company_id}/{file.id}{file_ext}"
|
||||
file.stored_filename = actual_stored
|
||||
await db.flush()
|
||||
|
||||
# 解析并写入 ParsedFileRecord
|
||||
parsed = _parse_csv(csv_content, bucket)
|
||||
for idx, rec in enumerate(parsed):
|
||||
pfr = ParsedFileRecord(
|
||||
company_id=file.company_id,
|
||||
task_id=file.task_id,
|
||||
uploaded_file_id=file.id,
|
||||
file_bucket=bucket,
|
||||
row_index=idx,
|
||||
raw_data=rec["raw_data"],
|
||||
normalized_data=rec["normalized_data"],
|
||||
employee_id=rec["employee_id"],
|
||||
employee_name=rec["employee_name"],
|
||||
)
|
||||
db.add(pfr)
|
||||
created += 1
|
||||
|
||||
print(f" 文件 {file.id}: bucket={bucket}, {len(parsed)} 行")
|
||||
|
||||
await db.commit()
|
||||
print(f"\n完成：创建 {created} 条 ParsedFileRecord")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(backfill())
|
||||
@@ -0,0 +1,27 @@
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
import asyncio
|
||||
from app.core.database import async_session_maker
|
||||
from sqlalchemy import text
|
||||
|
||||
async def check():
|
||||
async with async_session_maker() as db:
|
||||
r1 = await db.execute(text("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"))
|
||||
print('Tables:', [row[0] for row in r1.fetchall()])
|
||||
r2 = await db.execute(text("SELECT column_name FROM information_schema.columns WHERE table_name = 'companies' AND table_schema = 'public'"))
|
||||
print('companies cols:', [row[0] for row in r2.fetchall()])
|
||||
r3 = await db.execute(text("SELECT column_name FROM information_schema.columns WHERE table_name = 'reconciliation_tasks' AND table_schema = 'public'"))
|
||||
print('reconciliation_tasks cols:', [row[0] for row in r3.fetchall()])
|
||||
r4 = await db.execute(text("SELECT column_name FROM information_schema.columns WHERE table_name = 'uploaded_files' AND table_schema = 'public'"))
|
||||
print('uploaded_files cols:', [row[0] for row in r4.fetchall()])
|
||||
|
||||
r5 = await db.execute(text("SELECT id, name FROM companies LIMIT 5"))
|
||||
print('Companies:', r5.fetchall())
|
||||
r6 = await db.execute(text("SELECT id, company_id, name, status FROM reconciliation_tasks LIMIT 10"))
|
||||
print('Tasks:', r6.fetchall())
|
||||
r7 = await db.execute(text("SELECT id, task_id, company_id, file_type, parse_status, stored_filename FROM uploaded_files LIMIT 10"))
|
||||
print('UploadedFiles:', r7.fetchall())
|
||||
|
||||
asyncio.run(check())
|
||||
Reference in New Issue
Block a user