contract_repo.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215
  1. from sqlalchemy.ext.asyncio import AsyncSession
  2. from alien_store.db.models.contract_store import ContractStore
  3. import json
  4. from datetime import datetime, timedelta
  5. class ContractRepository:
  6. """合同数据访问层"""
  7. def __init__(self, db: AsyncSession):
  8. self.db = db
  9. async def get_by_store_id(self, store_id: int):
  10. """根据店铺id查询所有合同"""
  11. result = await self.db.execute(
  12. ContractStore.__table__.select().where(ContractStore.store_id == store_id)
  13. )
  14. # 返回列表[dict],避免 Pydantic 序列化 Row 对象出错
  15. return [dict(row) for row in result.mappings().all()]
  16. async def get_all(self):
  17. """查询所有合同"""
  18. result = await self.db.execute(ContractStore.__table__.select())
  19. return [dict(row) for row in result.mappings().all()]
  20. async def get_all_paged(self, page: int, page_size: int = 10):
  21. """分页查询所有合同,返回 (items, total)"""
  22. offset = (page - 1) * page_size
  23. # 查询总数
  24. from sqlalchemy import func, select
  25. count_result = await self.db.execute(
  26. select(func.count()).select_from(ContractStore.__table__)
  27. )
  28. total = count_result.scalar() or 0
  29. # 查询分页数据
  30. result = await self.db.execute(
  31. ContractStore.__table__.select().offset(offset).limit(page_size)
  32. )
  33. items = [dict(row) for row in result.mappings().all()]
  34. return items, total
  35. async def create(self, user_data):
  36. """创建未签署合同模板"""
  37. db_templates = ContractStore(
  38. store_id=user_data.store_id,
  39. merchant_name=user_data.merchant_name,
  40. business_segment=user_data.business_segment,
  41. contact_phone=user_data.contact_phone,
  42. contract_url=user_data.contract_url,
  43. seal_url='0.0',
  44. signing_status='未签署'
  45. )
  46. self.db.add(db_templates)
  47. await self.db.commit()
  48. await self.db.refresh(db_templates)
  49. return db_templates
  50. async def mark_signed_by_phone(self, contact_phone: str, sign_flow_id: str, signing_time: datetime | None = None, contract_download_url: str | None = None):
  51. """
  52. 根据手机号 + sign_flow_id 将合同标记为已签署,只更新匹配的合同项
  53. 当 is_master 为 1 时,更新签署状态和时间字段
  54. 同时写入签署/生效/到期时间(签署时间=T,生效=T+1天0点,失效=生效+365天)
  55. 同时更新 contract_download_url 到对应的字典中
  56. """
  57. result = await self.db.execute(
  58. ContractStore.__table__.select().where(ContractStore.contact_phone == contact_phone)
  59. )
  60. rows = result.mappings().all()
  61. updated = False
  62. for row in rows:
  63. contract_url_raw = row.get("contract_url")
  64. items = None
  65. if contract_url_raw:
  66. try:
  67. items = json.loads(contract_url_raw)
  68. except Exception:
  69. items = None
  70. changed = False
  71. matched_item = None
  72. if isinstance(items, list):
  73. for item in items:
  74. if item.get("sign_flow_id") == sign_flow_id:
  75. item["status"] = 1
  76. # 更新 contract_download_url
  77. if contract_download_url:
  78. item["contract_download_url"] = contract_download_url
  79. matched_item = item
  80. changed = True
  81. break
  82. # 只有当 is_master 为 1 时才更新时间字段
  83. if changed and matched_item and matched_item.get("is_master") == 1:
  84. # 时间处理
  85. signing_dt = signing_time
  86. effective_dt = expiry_dt = None
  87. if signing_dt:
  88. # effective_time 是 signing_time 第二天的 0 点
  89. effective_dt = (signing_dt + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
  90. expiry_dt = effective_dt + timedelta(days=365)
  91. # 更新 contract_url 中对应字典的时间字段
  92. matched_item["signing_time"] = signing_dt.strftime("%Y-%m-%d %H:%M:%S") if signing_dt else ""
  93. matched_item["effective_time"] = effective_dt.strftime("%Y-%m-%d %H:%M:%S") if effective_dt else ""
  94. matched_item["expiry_time"] = expiry_dt.strftime("%Y-%m-%d %H:%M:%S") if expiry_dt else ""
  95. await self.db.execute(
  96. ContractStore.__table__.update()
  97. .where(ContractStore.id == row["id"])
  98. .values(
  99. signing_status="已签署",
  100. contract_url=json.dumps(items, ensure_ascii=False) if items else contract_url_raw,
  101. signing_time=signing_dt,
  102. effective_time=effective_dt,
  103. expiry_time=expiry_dt,
  104. )
  105. )
  106. updated = True
  107. elif changed:
  108. # is_master 不为 1 时,只更新 status,不更新时间字段
  109. await self.db.execute(
  110. ContractStore.__table__.update()
  111. .where(ContractStore.id == row["id"])
  112. .values(
  113. contract_url=json.dumps(items, ensure_ascii=False) if items else contract_url_raw,
  114. )
  115. )
  116. updated = True
  117. if updated:
  118. await self.db.commit()
  119. return updated
  120. async def update_sign_url(self, contact_phone: str, sign_flow_id: str, sign_url: str):
  121. """
  122. 根据手机号 + sign_flow_id 更新 contract_url 列表中对应项的 sign_url
  123. """
  124. result = await self.db.execute(
  125. ContractStore.__table__.select().where(ContractStore.contact_phone == contact_phone)
  126. )
  127. rows = result.mappings().all()
  128. updated = False
  129. for row in rows:
  130. contract_url_raw = row.get("contract_url")
  131. if not contract_url_raw:
  132. continue
  133. try:
  134. items = json.loads(contract_url_raw)
  135. except Exception:
  136. items = None
  137. if not isinstance(items, list):
  138. continue
  139. changed = False
  140. for item in items:
  141. if item.get("sign_flow_id") == sign_flow_id:
  142. item["sign_url"] = sign_url
  143. changed = True
  144. if changed:
  145. await self.db.execute(
  146. ContractStore.__table__.update()
  147. .where(ContractStore.id == row["id"])
  148. .values(contract_url=json.dumps(items, ensure_ascii=False))
  149. )
  150. updated = True
  151. if updated:
  152. await self.db.commit()
  153. return updated
  154. async def append_contract_url(self, templates_data, contract_item: dict):
  155. """
  156. 根据手机号,向 contract_url(JSON 列表)追加新的合同信息;
  157. 若手机号不存在,则创建新记录。
  158. """
  159. contact_phone = getattr(templates_data, "contact_phone", None)
  160. result = await self.db.execute(
  161. ContractStore.__table__.select().where(ContractStore.contact_phone == contact_phone)
  162. )
  163. rows = result.mappings().all()
  164. updated = False
  165. if rows:
  166. for row in rows:
  167. contract_url_raw = row.get("contract_url")
  168. try:
  169. items = json.loads(contract_url_raw) if contract_url_raw else []
  170. except Exception:
  171. items = []
  172. if not isinstance(items, list):
  173. items = []
  174. items.append(contract_item)
  175. await self.db.execute(
  176. ContractStore.__table__.update()
  177. .where(ContractStore.id == row["id"])
  178. .values(contract_url=json.dumps(items, ensure_ascii=False))
  179. )
  180. updated = True
  181. if updated:
  182. await self.db.commit()
  183. return updated
  184. # 未找到则创建新记录
  185. new_record = ContractStore(
  186. store_id=getattr(templates_data, "store_id", None),
  187. business_segment=getattr(templates_data, "business_segment", None),
  188. merchant_name=getattr(templates_data, "merchant_name", None),
  189. contact_phone=contact_phone,
  190. contract_url=json.dumps([contract_item], ensure_ascii=False),
  191. seal_url='0.0',
  192. signing_status='未签署'
  193. )
  194. self.db.add(new_record)
  195. await self.db.commit()
  196. await self.db.refresh(new_record)
  197. return True