JiuContinent/company-ai-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
company-ai-platform/app/modules/legacy_mysql/intasect.py
JiuContinent 9cf7c44393 ``` 
feat: 添加生命周期报告和AI规则管理功能

- 在Dockerfile中添加pillow依赖包用于图像处理
- 实现生命周期报告调度任务，支持日报和周报两种类型
- 新增TASK_RUN_LIFECYCLE任务常量和相关配置选项
- 扩展AI Agent服务以支持用户规则，并在分析时应用规则
- 添加AI用户规则创建、更新和查询接口
- 增加项目生命周期和财务需求分析技能
- 扩展现有模型以支持更完整的业务数据字段
- 实现飞书图片上传功能用于报告展示
```
2026-07-12 17:44:49 +08:00

1144 lines
46 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from typing import Any
from fastapi import HTTPException, status
from sqlalchemy import Engine, Select, select, text, update
from sqlalchemy.orm import Session
from app.core.config import get_settings
from app.core.database import legacy_engine
from app.core.utils.time import utc_now
from app.modules.business.constants import (
CashFlowDirection,
CashFlowType,
DataQualityStatus,
RiskLevel,
SourceSystem,
StatusValue,
)
from app.modules.business.models import (
AttendanceRecord,
Employee,
Project,
ProjectCashFlow,
ProjectContract,
ProjectMember,
ProjectMilestone,
RiskEvent,
SourceSyncCursor,
WorkReport,
WorkTask,
)
from app.modules.legacy_mysql.constants import LegacyQueryError
INTASECT_SOURCE = "intasect"
DEFAULT_BATCH_SIZE = 500
EPOCH = datetime(1970, 1, 1)
PROJECT_SQL = """
SELECT CAST(p.id AS CHAR) AS source_key, p.id AS source_id,
p.business_code, p.pro_sn, p.name, p.mgr_deptid, p.mgr_deptname,
p.mgr_user_id, p.mgr_user_name, p.project_stage,
COALESCE(stage_dict.dict_label, p.project_stage) AS stage_label,
p.archive_flag, p.create_time AS source_created_at,
COALESCE(p.update_time, p.create_time) AS source_updated_at,
p.contract_date, p.project_done_date, p.project_information,
p.contract_money, p.project_invest_amount
FROM t_project p
LEFT JOIN sys_dict_data stage_dict
ON stage_dict.dict_type = 'project_stage' AND stage_dict.dict_value = p.project_stage
WHERE p.del_flag = '0' AND p.id > :after_key
ORDER BY p.id
LIMIT :limit
"""
EMPLOYEE_SQL = """
SELECT CAST(u.user_id AS CHAR) AS source_key, u.user_id AS source_id,
u.user_name AS employee_name, u.dept_id, d.dept_name,
u.status AS employment_status, u.ding_id, du.title,
du.hired_date, u.create_time AS source_created_at,
COALESCE(u.update_time, u.create_time, du.update_time, du.create_time) AS source_updated_at
FROM sys_user u
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
LEFT JOIN ding_user du ON du.userid = u.ding_id
WHERE u.del_flag = '0' AND u.user_id > :after_key
ORDER BY u.user_id
LIMIT :limit
"""
PROJECT_MEMBER_SQL = """
SELECT CAST(pu.id AS CHAR) AS source_key, pu.id AS source_id,
pu.project_id, pu.user_id, post.post_name AS role_name,
pu.leader, pu.resident, pu.work_mode, pu.work_time, pu.worknum_rate,
pu.create_time AS source_created_at, pu.create_time AS source_updated_at
FROM t_project_user pu
JOIN t_project p ON p.id = pu.project_id AND p.del_flag = '0'
JOIN sys_user u ON u.user_id = pu.user_id AND u.del_flag = '0'
LEFT JOIN sys_post post ON post.post_id = pu.post_id
WHERE pu.id > :after_key
ORDER BY pu.id
LIMIT :limit
"""
TASK_SQL = """
SELECT CAST(t.id AS CHAR) AS source_key, t.id AS source_id,
t.task_name, t.task_project_id, p.del_flag AS project_del_flag,
t.task_assignee, assignee.user_name AS assignee_name,
t.task_status, t.task_level, t.task_start_time, t.task_end_time,
t.task_done_time, t.task_content, t.task_pause,
t.create_time AS source_created_at,
COALESCE(t.update_time, t.create_time) AS source_updated_at
FROM dt_task t
LEFT JOIN t_project p ON p.id = t.task_project_id
LEFT JOIN sys_user assignee ON CAST(assignee.user_id AS CHAR) = t.task_assignee
WHERE t.id > :after_key
ORDER BY t.id
LIMIT :limit
"""
MILESTONE_SQL = """
SELECT CONCAT(LPAD(ps.project_id, 20, '0'), ':', ps.stage_id, ':', COALESCE(ps.sub_proid, '')) AS source_key,
ps.project_id, ps.stage_id, COALESCE(ls.stage_name, ps.stage_id) AS stage_name,
MIN(ps.plan_start) AS plan_start, MAX(ps.plan_end) AS plan_end,
MAX(ps.is_end) AS is_end, MAX(ps.is_overtime) AS is_overtime,
COALESCE(activity.activity_total, 0) AS activity_total,
COALESCE(activity.activity_completed, 0) AS activity_completed,
MAX(COALESCE(ps.update_time, ps.create_time)) AS source_updated_at
FROM t_project_stone ps
JOIN t_project p ON p.id = ps.project_id AND p.del_flag = '0'
LEFT JOIN t_lcb_stage ls ON ls.stage_id = ps.stage_id
LEFT JOIN (
SELECT project_id, stage_id, COALESCE(sub_proid, '') AS sub_proid,
COUNT(*) AS activity_total,
SUM(finish_flag = 'Y') AS activity_completed
FROM t_project_activity
GROUP BY project_id, stage_id, COALESCE(sub_proid, '')
) activity ON activity.project_id = ps.project_id
AND activity.stage_id = ps.stage_id
AND activity.sub_proid = COALESCE(ps.sub_proid, '')
WHERE CONCAT(LPAD(ps.project_id, 20, '0'), ':', ps.stage_id, ':', COALESCE(ps.sub_proid, '')) > :after_key
GROUP BY ps.project_id, ps.stage_id, COALESCE(ps.sub_proid, ''),
COALESCE(ls.stage_name, ps.stage_id),
activity.activity_total, activity.activity_completed
ORDER BY source_key
LIMIT :limit
"""
PROJECT_EVENT_SQL = """
SELECT e.id AS source_key, e.id AS source_id, e.project_id,
e.project_status, e.status AS event_status, e.need_help,
e.plan_date, e.end_date, e.event, e.event_description,
COALESCE(e.update_time, e.create_time) AS source_updated_at
FROM t_project_event e
JOIN t_project p ON p.id = e.project_id AND p.del_flag = '0'
WHERE e.id > :after_key AND (e.status = 'YCQ' OR e.need_help = 'Y')
ORDER BY e.id
LIMIT :limit
"""
DING_ATTENDANCE_SQL = """
SELECT changed.source_key, a.userid, a.work_date, u.user_id, u.user_name, d.dept_name,
MIN(CASE WHEN a.check_type = 'OnDuty' THEN a.user_check_time END) AS check_in_at,
MAX(CASE WHEN a.check_type = 'OffDuty' THEN a.user_check_time END) AS check_out_at,
GROUP_CONCAT(DISTINCT a.time_result ORDER BY a.time_result) AS time_results,
GROUP_CONCAT(DISTINCT a.location_result ORDER BY a.location_result) AS location_results,
changed.source_updated_at
FROM ding_user_attendance a
JOIN (
SELECT attendance_source.userid, attendance_source.work_date,
CONCAT(attendance_source.userid, ':', DATE_FORMAT(attendance_source.work_date, '%Y-%m-%d')) AS source_key,
MAX(COALESCE(attendance_source.update_time, attendance_source.create_time, attendance_source.user_check_time)) AS source_updated_at
FROM ding_user_attendance attendance_source
JOIN sys_user changed_user
ON changed_user.ding_id = attendance_source.userid AND changed_user.del_flag = '0'
WHERE attendance_source.work_date IS NOT NULL
GROUP BY attendance_source.userid, attendance_source.work_date
HAVING source_updated_at > :watermark_at
OR (source_updated_at = :watermark_at AND source_key > :after_key)
ORDER BY source_updated_at, source_key
LIMIT :limit
) changed ON changed.userid = a.userid AND changed.work_date = a.work_date
JOIN sys_user u ON u.ding_id = a.userid AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
GROUP BY changed.source_key, changed.source_updated_at,
a.userid, a.work_date, u.user_id, u.user_name, d.dept_name
ORDER BY changed.source_updated_at, changed.source_key
"""
DING_ATTENDANCE_FULL_SQL = """
SELECT CONCAT(a.userid, ':', DATE_FORMAT(a.work_date, '%Y-%m-%d')) AS source_key,
a.userid, a.work_date, u.user_id, u.user_name, d.dept_name,
MIN(CASE WHEN a.check_type = 'OnDuty' THEN a.user_check_time END) AS check_in_at,
MAX(CASE WHEN a.check_type = 'OffDuty' THEN a.user_check_time END) AS check_out_at,
GROUP_CONCAT(DISTINCT a.time_result ORDER BY a.time_result) AS time_results,
GROUP_CONCAT(DISTINCT a.location_result ORDER BY a.location_result) AS location_results,
MAX(COALESCE(a.update_time, a.create_time, a.user_check_time)) AS source_updated_at
FROM ding_user_attendance a
JOIN sys_user u ON u.ding_id = a.userid AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
WHERE a.work_date IS NOT NULL
AND (a.userid > :after_userid
OR (a.userid = :after_userid AND a.work_date > :after_date))
GROUP BY a.userid, a.work_date, u.user_id, u.user_name, d.dept_name
ORDER BY a.userid, a.work_date
LIMIT :limit
"""
PROJECT_ATTENDANCE_SQL = """
SELECT CAST(s.id AS CHAR) AS source_key, s.id AS source_id,
s.project_id, s.user_id, u.user_name, d.dept_name,
DATE(COALESCE(s.in_time, s.create_time)) AS work_date,
s.in_time AS check_in_at, s.out_time AS check_out_at,
s.is_delay, COALESCE(s.create_time, s.in_time, s.out_time) AS source_updated_at
FROM t_project_usersign s
JOIN t_project p ON p.id = s.project_id AND p.del_flag = '0'
JOIN sys_user u ON u.user_id = s.user_id AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
WHERE (COALESCE(s.create_time, s.in_time, s.out_time) > :watermark_at
OR (COALESCE(s.create_time, s.in_time, s.out_time) = :watermark_at AND s.id > :after_key))
ORDER BY source_updated_at, s.id
LIMIT :limit
"""
PROJECT_ATTENDANCE_FULL_SQL = """
SELECT CAST(s.id AS CHAR) AS source_key, s.id AS source_id,
s.project_id, s.user_id, u.user_name, d.dept_name,
DATE(COALESCE(s.in_time, s.create_time)) AS work_date,
s.in_time AS check_in_at, s.out_time AS check_out_at,
s.is_delay, COALESCE(s.create_time, s.in_time, s.out_time) AS source_updated_at
FROM t_project_usersign s
JOIN t_project p ON p.id = s.project_id AND p.del_flag = '0'
JOIN sys_user u ON u.user_id = s.user_id AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
WHERE s.id > :after_key
ORDER BY s.id
LIMIT :limit
"""
WORK_REPORT_SQL = """
SELECT r.id AS source_key, r.id AS source_id, r.user_id, u.user_name,
d.dept_name, r.date AS period_end, r.begin_date AS period_start,
r.type AS report_type, r.late, r.draft,
r.create_time AS source_created_at,
COALESCE(r.update_time, r.create_time) AS source_updated_at
FROM t_daily r
JOIN sys_user u ON u.user_id = r.user_id AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
WHERE (COALESCE(r.update_time, r.create_time) > :watermark_at
OR (COALESCE(r.update_time, r.create_time) = :watermark_at AND r.id > :after_key))
ORDER BY source_updated_at, r.id
LIMIT :limit
"""
WORK_REPORT_FULL_SQL = """
SELECT r.id AS source_key, r.id AS source_id, r.user_id, u.user_name,
d.dept_name, r.date AS period_end, r.begin_date AS period_start,
r.type AS report_type, r.late, r.draft,
r.create_time AS source_created_at,
COALESCE(r.update_time, r.create_time) AS source_updated_at
FROM t_daily r
JOIN sys_user u ON u.user_id = r.user_id AND u.del_flag = '0'
LEFT JOIN sys_dept d ON d.dept_id = u.dept_id
WHERE r.id > :after_key
ORDER BY r.id
LIMIT :limit
"""
CONTRACT_SQL = """
SELECT CAST(c.id AS CHAR) AS source_key, c.id AS source_id,
c.project_id, p.id AS linked_project_id, p.del_flag AS project_del_flag,
c.type AS contract_type_code,
COALESCE(contract_type.dict_label, c.type) AS contract_type_label,
c.contract_amount, c.contract_date, c.date_start, c.date_end,
c.pay_way AS invoice_type_code, c.create_time AS source_created_at
FROM t_contract c
LEFT JOIN t_project p ON p.id = c.project_id
LEFT JOIN sys_dict_data contract_type
ON contract_type.dict_type = 'contract_type' AND contract_type.dict_value = c.type
WHERE c.id > :after_key
ORDER BY c.id
LIMIT :limit
"""
CONTRACT_RECEIVABLE_SQL = """
SELECT CAST(cp.id AS CHAR) AS source_key, cp.id AS source_id,
cp.contract_id AS source_contract_id, c.id AS linked_contract_id,
c.project_id, p.id AS linked_project_id, p.del_flag AS project_del_flag,
cp.pay_node AS category_code,
COALESCE(pay_node.dict_label, cp.pay_node) AS category_label,
cp.pay_num AS planned_amount, cp.real_num AS actual_amount,
COALESCE(cp.plan_date, cp.pay_date) AS planned_date, cp.real_date AS actual_date,
cp.status AS payment_status, cp.fapiao AS invoice_status,
cp.create_time AS source_created_at
FROM t_contract_pay cp
LEFT JOIN t_contract c ON c.id = cp.contract_id
LEFT JOIN t_project p ON p.id = c.project_id
LEFT JOIN sys_dict_data pay_node
ON pay_node.dict_type = 'pay_node' AND pay_node.dict_value = cp.pay_node
WHERE cp.id > :after_key
ORDER BY cp.id
LIMIT :limit
"""
PROJECT_FUND_SQL = """
SELECT CAST(f.id AS CHAR) AS source_key, f.id AS source_id,
f.project_id, p.id AS linked_project_id, p.del_flag AS project_del_flag,
f.cost_class, cost_class.dict_label AS cost_class_label,
f.cost_type AS category_code,
COALESCE(cost_type.dict_label, f.cost_type) AS category_label,
f.cost_amount AS planned_amount, f.approval_status, f.confirm_status,
f.approval_time, f.trade_time,
f.create_time AS source_created_at,
COALESCE(f.update_time, f.create_time) AS source_updated_at
FROM dt_bid_fund_detail f
LEFT JOIN t_project p ON p.id = f.project_id
LEFT JOIN sys_dict_data cost_class
ON cost_class.dict_type = 'pms_cost_class' AND cost_class.dict_value = f.cost_class
LEFT JOIN sys_dict_data cost_type
ON cost_type.dict_type = 'pms_cost_type' AND cost_type.dict_value = f.cost_type
WHERE f.id > :after_key
ORDER BY f.id
LIMIT :limit
"""
CORE_FULL_DATASETS = (
"projects",
"employees",
"project_members",
"tasks",
"milestones",
"project_events",
)
FINANCE_DATASETS = ("contracts", "contract_receivables", "project_funds")
FULL_DATASETS = (*CORE_FULL_DATASETS, *FINANCE_DATASETS)
INCREMENTAL_DATASETS = ("company_attendance", "project_attendance", "work_reports")
DATASET_SQL = {
"projects": PROJECT_SQL,
"employees": EMPLOYEE_SQL,
"project_members": PROJECT_MEMBER_SQL,
"tasks": TASK_SQL,
"milestones": MILESTONE_SQL,
"project_events": PROJECT_EVENT_SQL,
"company_attendance": DING_ATTENDANCE_SQL,
"project_attendance": PROJECT_ATTENDANCE_SQL,
"work_reports": WORK_REPORT_SQL,
"contracts": CONTRACT_SQL,
"contract_receivables": CONTRACT_RECEIVABLE_SQL,
"project_funds": PROJECT_FUND_SQL,
}
class IntasectSourceRepository:
def __init__(self, engine: Engine | None = None):
self.engine = engine or legacy_engine
if self.engine is None:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=LegacyQueryError.DATABASE_NOT_CONFIGURED,
)
def fetch_page(
self,
dataset: str,
after_key: str,
watermark_at: datetime,
limit: int = DEFAULT_BATCH_SIZE,
) -> list[dict[str, Any]]:
full_sql = {
"company_attendance": DING_ATTENDANCE_FULL_SQL,
"project_attendance": PROJECT_ATTENDANCE_FULL_SQL,
"work_reports": WORK_REPORT_FULL_SQL,
}
sql = (
full_sql.get(dataset, DATASET_SQL[dataset])
if watermark_at == EPOCH
else DATASET_SQL[dataset]
)
after_userid = ""
after_date = date(1900, 1, 1)
if dataset == "company_attendance" and after_key:
after_userid, raw_date = after_key.rsplit(":", 1)
after_date = date.fromisoformat(raw_date)
params: dict[str, Any] = {
"after_key": int(after_key or 0)
if dataset
in {
"projects",
"contracts",
"contract_receivables",
"project_funds",
"employees",
"project_members",
"tasks",
"project_attendance",
}
else after_key,
"watermark_at": watermark_at,
"after_userid": after_userid,
"after_date": after_date,
"limit": min(max(int(limit), 1), DEFAULT_BATCH_SIZE),
}
with self.engine.connect() as connection:
return [dict(row) for row in connection.execute(text(sql), params).mappings()]
class IntasectSyncService:
def __init__(self, db: Session, source: IntasectSourceRepository | None = None):
self.db = db
self.source = source
def sync_all(
self,
run_code: str,
force_full: bool = False,
batch_size: int = DEFAULT_BATCH_SIZE,
) -> dict[str, Any]:
results: dict[str, Any] = {}
full_datasets = list(CORE_FULL_DATASETS)
if get_settings().finance_needs_enabled:
full_datasets[1:1] = FINANCE_DATASETS
for dataset in (*full_datasets, *INCREMENTAL_DATASETS):
results[dataset] = self.sync_dataset(dataset, run_code, force_full, batch_size)
return results
def sync_dataset(
self,
dataset: str,
run_code: str,
force_full: bool = False,
batch_size: int = DEFAULT_BATCH_SIZE,
) -> dict[str, Any]:
cursor = self._cursor(dataset)
if self.source is None:
self.source = IntasectSourceRepository()
incremental = (
dataset in INCREMENTAL_DATASETS and cursor.watermark_at is not None and not force_full
)
watermark_at = cursor.watermark_at if incremental and cursor.watermark_at else EPOCH
after_key = cursor.watermark_key if incremental and cursor.watermark_key else ""
seen_at = utc_now()
cursor.status = StatusValue.RUNNING
cursor.last_run_code = run_code
cursor.error_message = None
self.db.commit()
processed = 0
latest_source_updated: datetime | None = None
latest_source_key = ""
try:
while True:
rows = self.source.fetch_page(dataset, after_key, watermark_at, batch_size)
if not rows:
break
self._apply_rows(dataset, rows, seen_at)
processed += len(rows)
last = rows[-1]
after_key = str(last["source_key"])
last_updated = _as_datetime(last.get("source_updated_at"))
if incremental and last_updated:
watermark_at = last_updated
for row in rows:
row_updated = _as_datetime(row.get("source_updated_at"))
row_key = str(row["source_key"])
if row_updated and (
latest_source_updated is None
or row_updated > latest_source_updated
or (row_updated == latest_source_updated and row_key > latest_source_key)
):
latest_source_updated = row_updated
latest_source_key = row_key
self.db.commit()
if len(rows) < batch_size:
break
if dataset in FULL_DATASETS:
self._deactivate_missing(dataset, seen_at)
cursor.status = StatusValue.COMPLETED
cursor.processed_count = processed
if dataset in INCREMENTAL_DATASETS and latest_source_updated:
cursor.watermark_at = latest_source_updated
cursor.watermark_key = latest_source_key
else:
cursor.watermark_at = seen_at
cursor.watermark_key = after_key
cursor.last_success_at = utc_now()
self.db.commit()
except Exception as exc:
self.db.rollback()
cursor = self._cursor(dataset)
cursor.status = "failed"
cursor.last_run_code = run_code
cursor.processed_count = processed
cursor.error_message = str(exc)[:2000]
self.db.commit()
raise
return {
"dataset": dataset,
"processed": processed,
"incremental": incremental,
"watermark_at": cursor.watermark_at.isoformat() if cursor.watermark_at else None,
"watermark_key": cursor.watermark_key,
}
def _cursor(self, dataset: str) -> SourceSyncCursor:
record = self.db.execute(
select(SourceSyncCursor).where(SourceSyncCursor.dataset == dataset)
).scalar_one_or_none()
if record is None:
record = SourceSyncCursor(dataset=dataset)
self.db.add(record)
self.db.commit()
self.db.refresh(record)
return record
def _apply_rows(self, dataset: str, rows: list[dict[str, Any]], seen_at: datetime) -> None:
handlers = {
"projects": self._upsert_projects,
"contracts": self._upsert_contracts,
"contract_receivables": self._upsert_contract_receivables,
"project_funds": self._upsert_project_funds,
"employees": self._upsert_employees,
"project_members": self._upsert_members,
"tasks": self._upsert_tasks,
"milestones": self._upsert_milestones,
"project_events": self._upsert_events,
"company_attendance": self._upsert_company_attendance,
"project_attendance": self._upsert_project_attendance,
"work_reports": self._upsert_work_reports,
}
handlers[dataset](rows, seen_at)
def _existing(self, model: type, external_ids: list[str]) -> dict[str, Any]:
statement: Select = select(model).where(
model.source_system == SourceSystem.LEGACY_MYSQL,
model.external_id.in_(external_ids),
)
return {str(item.external_id): item for item in self.db.execute(statement).scalars()}
def _upsert_projects(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
existing = self._existing(Project, [str(row["source_id"]) for row in rows])
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _project_payload(row, seen_at)
if record is None:
self.db.add(Project(**payload))
else:
_assign(record, payload)
def _upsert_employees(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
existing = self._existing(Employee, [str(row["source_id"]) for row in rows])
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _employee_payload(row, seen_at)
if record is None:
self.db.add(Employee(**payload))
else:
_assign(record, payload)
def _upsert_contracts(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
existing = self._existing(ProjectContract, [str(row["source_id"]) for row in rows])
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _contract_payload(row, seen_at)
if record is None:
self.db.add(ProjectContract(**payload))
else:
_assign(record, payload)
def _upsert_contract_receivables(
self, rows: list[dict[str, Any]], seen_at: datetime
) -> None:
ids = [f"receivable:{row['source_id']}" for row in rows]
existing = self._existing(ProjectCashFlow, ids)
for row in rows:
external_id = f"receivable:{row['source_id']}"
record = existing.get(external_id)
payload = _contract_receivable_payload(row, seen_at)
if record is None:
self.db.add(ProjectCashFlow(**payload))
else:
_assign(record, payload)
def _upsert_project_funds(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
ids = [f"fund:{row['source_id']}" for row in rows]
existing = self._existing(ProjectCashFlow, ids)
for row in rows:
external_id = f"fund:{row['source_id']}"
record = existing.get(external_id)
payload = _project_fund_payload(row, seen_at)
if record is None:
self.db.add(ProjectCashFlow(**payload))
else:
_assign(record, payload)
def _upsert_members(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
existing = self._existing(ProjectMember, [str(row["source_id"]) for row in rows])
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _member_payload(row, seen_at)
if record is None:
self.db.add(ProjectMember(**payload))
else:
_assign(record, payload)
def _upsert_tasks(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
existing = self._existing(WorkTask, [str(row["source_id"]) for row in rows])
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _task_payload(row, seen_at)
if record is None:
self.db.add(WorkTask(**payload))
else:
_assign(record, payload)
def _upsert_milestones(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
ids = [str(row["source_key"]) for row in rows]
existing = self._existing(ProjectMilestone, ids)
for row in rows:
external_id = str(row["source_key"])
record = existing.get(external_id)
payload = _milestone_payload(row, seen_at)
if record is None:
self.db.add(ProjectMilestone(**payload))
else:
_assign(record, payload)
def _upsert_events(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
codes = [f"INTASECT-EVENT-{row['source_id']}" for row in rows]
existing = {
item.code: item
for item in self.db.execute(
select(RiskEvent).where(RiskEvent.code.in_(codes))
).scalars()
}
for row in rows:
code = f"INTASECT-EVENT-{row['source_id']}"
record = existing.get(code)
payload = _event_payload(row, seen_at)
if record is None:
self.db.add(RiskEvent(**payload))
else:
_assign(record, payload)
def _upsert_company_attendance(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
ids = [f"ding:{row['source_key']}" for row in rows]
existing = self._existing(AttendanceRecord, ids)
for row in rows:
external_id = f"ding:{row['source_key']}"
record = existing.get(external_id)
payload = _company_attendance_payload(row, seen_at)
if record is None:
self.db.add(AttendanceRecord(**payload))
else:
_assign(record, payload)
def _upsert_project_attendance(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
ids = [f"project:{row['source_id']}" for row in rows]
existing = self._existing(AttendanceRecord, ids)
for row in rows:
external_id = f"project:{row['source_id']}"
record = existing.get(external_id)
payload = _project_attendance_payload(row, seen_at)
if record is None:
self.db.add(AttendanceRecord(**payload))
else:
_assign(record, payload)
def _upsert_work_reports(self, rows: list[dict[str, Any]], seen_at: datetime) -> None:
ids = [str(row["source_id"]) for row in rows]
existing = self._existing(WorkReport, ids)
for row in rows:
external_id = str(row["source_id"])
record = existing.get(external_id)
payload = _work_report_payload(row, seen_at)
if record is None:
self.db.add(WorkReport(**payload))
else:
_assign(record, payload)
def _deactivate_missing(self, dataset: str, seen_at: datetime) -> None:
if dataset == "project_events":
self.db.execute(
update(RiskEvent)
.where(
RiskEvent.source_domain == "intasect_project_event",
RiskEvent.updated_at < seen_at,
)
.values(status=StatusValue.RESOLVED)
)
self.db.commit()
return
targets: dict[str, tuple[type, list[Any]]] = {
"projects": (Project, []),
"contracts": (ProjectContract, []),
"contract_receivables": (
ProjectCashFlow,
[ProjectCashFlow.flow_type == CashFlowType.CONTRACT_RECEIVABLE],
),
"project_funds": (
ProjectCashFlow,
[ProjectCashFlow.flow_type == CashFlowType.PROJECT_FUND],
),
"employees": (Employee, []),
"project_members": (ProjectMember, []),
"tasks": (WorkTask, []),
"milestones": (ProjectMilestone, []),
}
target = targets.get(dataset)
if target is None:
return
model, extra_conditions = target
conditions = [
model.source_system == SourceSystem.LEGACY_MYSQL,
(model.last_seen_at.is_(None) | (model.last_seen_at < seen_at)),
*extra_conditions,
]
self.db.execute(update(model).where(*conditions).values(is_active=False))
self.db.commit()
def _project_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
archived = str(row.get("archive_flag") or "0") == "2"
stage = _text(row.get("project_stage"))
completed = archived or stage == "XMWJ"
progress_by_stage = {"JD25": 25, "JD50": 50, "JD75": 75, "XMWJ": 100}
return {
"code": f"INTASECT-PROJECT-{external_id}",
"display_code": _text(row.get("business_code")) or _text(row.get("pro_sn")),
"name": _text(row.get("name")) or f"项目 {external_id}",
"owner": _text(row.get("mgr_user_name")),
"owner_employee_code": (
f"INTASECT-EMPLOYEE-{row['mgr_user_id']}" if row.get("mgr_user_id") else None
),
"department_code": _text(row.get("mgr_deptid")),
"department_name": _text(row.get("mgr_deptname")),
"status": StatusValue.COMPLETED if completed else StatusValue.RUNNING,
"progress_percent": progress_by_stage.get(stage or "", 0),
"source_contract_amount": _scaled_amount(row.get("contract_money"), Decimal("10000")),
"source_project_investment_amount": _scaled_amount(
row.get("project_invest_amount"), Decimal("10000")
),
"start_date": _as_date(row.get("contract_date")),
"due_date": None,
"description": _text(row.get("project_information")),
"source_stage": stage,
"source_stage_label": _text(row.get("stage_label")) or stage,
"source_archived": archived,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _contract_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
quality = _project_link_quality(row)
project_id = row.get("project_id")
return {
"code": f"INTASECT-CONTRACT-{external_id}",
"project_code": f"INTASECT-PROJECT-{project_id}" if project_id is not None else None,
"contract_type_code": _text(row.get("contract_type_code")),
"contract_type_label": _text(row.get("contract_type_label")),
"amount": _as_decimal(row.get("contract_amount")) or Decimal("0"),
"signed_date": _as_date(row.get("contract_date")),
"start_date": _as_date(row.get("date_start")),
"end_date": _as_date(row.get("date_end")),
"invoice_type_code": _text(row.get("invoice_type_code")),
"data_quality_status": quality,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _contract_receivable_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = f"receivable:{row['source_id']}"
actual = _as_decimal(row.get("actual_amount"))
payment_status = _text(row.get("payment_status"))
if row.get("linked_contract_id") is None:
quality = DataQualityStatus.ORPHAN_CONTRACT
else:
quality = _project_link_quality(row)
if quality == DataQualityStatus.VALID and payment_status == "Y" and not actual:
quality = DataQualityStatus.PAID_AMOUNT_MISSING
elif quality == DataQualityStatus.VALID and payment_status == "N" and actual and actual > 0:
quality = DataQualityStatus.STATUS_AMOUNT_MISMATCH
project_id = row.get("project_id")
contract_id = row.get("source_contract_id")
return {
"code": f"INTASECT-RECEIVABLE-{row['source_id']}",
"project_code": f"INTASECT-PROJECT-{project_id}" if project_id is not None else None,
"contract_code": (
f"INTASECT-CONTRACT-{contract_id}" if row.get("linked_contract_id") is not None else None
),
"flow_type": CashFlowType.CONTRACT_RECEIVABLE,
"direction": CashFlowDirection.INFLOW,
"category_code": _text(row.get("category_code")),
"category_label": _text(row.get("category_label")),
"planned_amount": _as_decimal(row.get("planned_amount")) or Decimal("0"),
"actual_amount": actual,
"planned_date": _as_date(row.get("planned_date")),
"actual_date": _as_date(row.get("actual_date")),
"payment_status": payment_status,
"invoice_status": _text(row.get("invoice_status")),
"approval_status": None,
"confirmation_status": None,
"data_quality_status": quality,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"source_updated_at": None,
"last_seen_at": seen_at,
"is_active": True,
}
def _project_fund_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = f"fund:{row['source_id']}"
amount = _as_decimal(row.get("planned_amount")) or Decimal("0")
confirmed = _text(row.get("approval_status")) == "1" and _text(
row.get("confirm_status")
) == "Y"
project_id = row.get("project_id")
return {
"code": f"INTASECT-FUND-{row['source_id']}",
"project_code": f"INTASECT-PROJECT-{project_id}" if project_id is not None else None,
"contract_code": None,
"flow_type": CashFlowType.PROJECT_FUND,
"direction": (
CashFlowDirection.INFLOW
if _text(row.get("cost_class")) == "1"
else CashFlowDirection.OUTFLOW
),
"category_code": _text(row.get("category_code")),
"category_label": _text(row.get("category_label")),
"planned_amount": amount,
"actual_amount": amount if confirmed else None,
"planned_date": None,
"actual_date": _as_date(row.get("trade_time")) if confirmed else None,
"payment_status": None,
"invoice_status": None,
"approval_status": _text(row.get("approval_status")),
"confirmation_status": _text(row.get("confirm_status")),
"data_quality_status": _project_link_quality(row),
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _employee_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
active = str(row.get("employment_status") or "0") == "0"
hired_at = None
if row.get("hired_date"):
try:
hired_at = datetime.fromtimestamp(int(row["hired_date"]) / 1000)
except (TypeError, ValueError, OSError):
hired_at = None
return {
"code": f"INTASECT-EMPLOYEE-{external_id}",
"name": _text(row.get("employee_name")) or f"员工 {external_id}",
"department_code": _text(row.get("dept_id")),
"department_name": _text(row.get("dept_name")),
"title": _text(row.get("title")),
"employment_status": "在职" if active else "离职",
"hired_at": hired_at,
"ding_user_id": _text(row.get("ding_id")),
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": active,
}
def _member_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
return {
"code": f"INTASECT-MEMBER-{external_id}",
"project_code": f"INTASECT-PROJECT-{row['project_id']}",
"employee_code": f"INTASECT-EMPLOYEE-{row['user_id']}",
"role_name": _text(row.get("role_name")),
"is_leader": str(row.get("leader") or "N").upper() == "Y",
"is_resident": str(row.get("resident") or "N").upper() == "Y",
"work_mode": _text(row.get("work_mode")),
"planned_days": _as_int(row.get("work_time")),
"workload_percent": _as_int(row.get("worknum_rate")),
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _task_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
completed = str(row.get("task_status")) == "1"
project_id = row.get("task_project_id") if str(row.get("project_del_flag")) == "0" else None
employee_id = _numeric_text(row.get("task_assignee"))
return {
"code": f"INTASECT-TASK-{external_id}",
"title": _text(row.get("task_name")) or f"任务 {external_id}",
"project_code": f"INTASECT-PROJECT-{project_id}" if project_id else None,
"owner": _text(row.get("assignee_name")),
"employee_code": f"INTASECT-EMPLOYEE-{employee_id}" if employee_id else None,
"status": StatusValue.COMPLETED if completed else StatusValue.RUNNING,
"priority": _task_priority(row.get("task_level")),
"due_date": _as_date(row.get("task_end_time")),
"completed_at": _as_datetime(row.get("task_done_time")),
"blocker": "任务已挂起" if _as_int(row.get("task_pause")) else None,
"description": _text(row.get("task_content")),
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_created_at": _as_datetime(row.get("source_created_at")),
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _milestone_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_key"])
ended = str(row.get("is_end") or "0") == "1"
plan_end = _as_date(row.get("plan_end"))
overdue = str(row.get("is_overtime") or "0") == "1" or (
bool(plan_end)
and plan_end < date.today()
and not ended
and str(row.get("is_end") or "0") != "2"
)
return {
"code": f"INTASECT-MILESTONE-{external_id}",
"project_code": f"INTASECT-PROJECT-{row['project_id']}",
"source_stage_id": str(row["stage_id"]),
"stage_name": _text(row.get("stage_name")) or str(row["stage_id"]),
"plan_start": _as_date(row.get("plan_start")),
"plan_end": plan_end,
"status": StatusValue.COMPLETED if ended else StatusValue.RUNNING,
"is_overdue": overdue,
"activity_total": _as_int(row.get("activity_total")) or 0,
"activity_completed": _as_int(row.get("activity_completed")) or 0,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _event_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
source_id = str(row["source_id"])
need_help = str(row.get("need_help") or "N").upper() == "Y"
event_status = str(row.get("event_status") or "")
return {
"code": f"INTASECT-EVENT-{source_id}",
"title": _text(row.get("event"))
or _text(row.get("event_description"))
or f"项目重大事项 {source_id}",
"risk_type": "project_event",
"risk_level": RiskLevel.HIGH if need_help else RiskLevel.MEDIUM,
"status": StatusValue.RESOLVED if event_status == "YWJ" else StatusValue.OPEN,
"source_domain": "intasect_project_event",
"source_record_id": source_id,
"project_code": f"INTASECT-PROJECT-{row['project_id']}",
"due_date": _as_date(row.get("plan_date")),
"description": _text(row.get("event_description")),
"evidence": {
"source_status": event_status,
"project_stage": _text(row.get("project_status")),
"need_help": need_help,
"source_updated_at": (
_as_datetime(row.get("source_updated_at")).isoformat()
if _as_datetime(row.get("source_updated_at"))
else None
),
"last_seen_at": seen_at.isoformat(),
},
}
def _company_attendance_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = f"ding:{row['source_key']}"
source_status = _text(row.get("time_results"))
return {
"code": f"INTASECT-ATTENDANCE-{row['user_id']}-{row['work_date']}",
"employee_name": _text(row.get("user_name")) or f"员工 {row['user_id']}",
"employee_id": f"INTASECT-EMPLOYEE-{row['user_id']}",
"department": _text(row.get("dept_name")),
"work_date": _as_date(row.get("work_date")),
"check_in_at": _as_datetime(row.get("check_in_at")),
"check_out_at": _as_datetime(row.get("check_out_at")),
"status": _attendance_status(source_status),
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"attendance_scope": "company",
"source_status": source_status,
"source_location_status": _text(row.get("location_results")),
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _project_attendance_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
source_id = str(row["source_id"])
delayed = str(row.get("is_delay") or "N").upper() == "Y"
return {
"code": f"INTASECT-PROJECT-SIGN-{source_id}",
"employee_name": _text(row.get("user_name")) or f"员工 {row['user_id']}",
"employee_id": f"INTASECT-EMPLOYEE-{row['user_id']}",
"department": _text(row.get("dept_name")),
"project_code": f"INTASECT-PROJECT-{row['project_id']}",
"work_date": _as_date(row.get("work_date")),
"check_in_at": _as_datetime(row.get("check_in_at")),
"check_out_at": _as_datetime(row.get("check_out_at")),
"status": StatusValue.LATE if delayed else StatusValue.NORMAL_CN,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": f"project:{source_id}",
"attendance_scope": "project",
"source_status": "delayed" if delayed else "normal",
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _work_report_payload(row: dict[str, Any], seen_at: datetime) -> dict[str, Any]:
external_id = str(row["source_id"])
period_end = _as_date(row.get("period_end")) or date.today()
period_start = _as_date(row.get("period_start")) or period_end
report_type = _text(row.get("report_type")) or "daily"
return {
"code": f"INTASECT-WORK-REPORT-{external_id}",
"report_type": report_type,
"title": f"{report_type}-{period_end.isoformat()}",
"reporter": _text(row.get("user_name")) or f"员工 {row['user_id']}",
"employee_code": f"INTASECT-EMPLOYEE-{row['user_id']}",
"department": _text(row.get("dept_name")),
"period_start": period_start,
"period_end": period_end,
"content": "",
"status": StatusValue.GENERATED,
"source_system": SourceSystem.LEGACY_MYSQL,
"external_id": external_id,
"is_late": str(row.get("late") or "N").upper() == "Y",
"is_draft": str(row.get("draft") or "N").upper() != "N",
"source_updated_at": _as_datetime(row.get("source_updated_at")),
"last_seen_at": seen_at,
"is_active": True,
}
def _attendance_status(source_status: str | None) -> str:
statuses = set((source_status or "").split(","))
if "Absenteeism" in statuses:
return StatusValue.ABSENT
if "NotSigned" in statuses:
return StatusValue.MISSING_PUNCH
if "Late" in statuses:
return StatusValue.LATE
if "Early" in statuses:
return StatusValue.LEAVE_EARLY
return StatusValue.NORMAL_CN
def _task_priority(value: Any) -> str:
return {"0": "P1", "1": "P2", "2": "P3"}.get(str(value), "P2")
def _assign(record: Any, payload: dict[str, Any]) -> None:
for key, value in payload.items():
setattr(record, key, value)
def _as_datetime(value: Any) -> datetime | None:
if isinstance(value, datetime):
return value
if isinstance(value, date):
return datetime.combine(value, datetime.min.time())
if value:
try:
return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError:
return None
return None
def _as_date(value: Any) -> date | None:
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
if value:
text_value = str(value).strip()[:10]
try:
return date.fromisoformat(text_value)
except ValueError:
return None
return None
def _as_int(value: Any) -> int | None:
if value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _as_decimal(value: Any) -> Decimal | None:
if value is None or value == "":
return None
try:
return Decimal(str(value))
except (InvalidOperation, ValueError, TypeError):
return None
def _scaled_amount(value: Any, multiplier: Decimal) -> Decimal | None:
amount = _as_decimal(value)
return amount * multiplier if amount is not None else None
def _project_link_quality(row: dict[str, Any]) -> DataQualityStatus:
if row.get("linked_project_id") is None:
return DataQualityStatus.ORPHAN_PROJECT
if str(row.get("project_del_flag") or "0") != "0":
return DataQualityStatus.DELETED_PROJECT
return DataQualityStatus.VALID
def _numeric_text(value: Any) -> str | None:
text_value = _text(value)
return text_value if text_value and text_value.isdigit() else None
def _text(value: Any) -> str | None:
if value is None:
return None
text_value = str(value).strip()
return text_value or None
Reference in New Issue View Git Blame Copy Permalink