# -*- coding: utf-8 -*- from datetime import datetime from decimal import Decimal, ROUND_HALF_UP from typing import Any, Dict, List, Optional from utils.client import SQLiteClient class MasterData(SQLiteClient): """主数据""" def __init__(self): """ 初始化主数据 """ # 初始化SQLite客户端 super().__init__(database="database.db") try: with self: # 初始化团单表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS group_policies ( --团单唯一标识 guid TEXT PRIMARY KEY, --团单号 group_policy TEXT NOT NULL, --保险分公司名称 insurer_company TEXT NOT NULL, --保险起期 commencement_date TEXT NOT NULL, --保险止期 termination_date TEXT NOT NULL ) """ ) # 初始化个单表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS person_policies ( --个单唯一标识 guid TEXT PRIMARY KEY, --个单号 person_policy TEXT NOT NULL, --保险起期 commencement_date TEXT NOT NULL, --保险止期 termination_date TEXT NOT NULL, --团单唯一标识，用于联查团案 group_policy_guid TEXT NOT NULL ) """ ) # 初始化被保险人表，保司推送赔案时，一般无团单号，需先根据保险分公司名称、被保险人姓名、证件类型和证件号码查询被保人，再在票据理算时根据事故起期确定个单和相应责任 self._execute( sql=""" CREATE TABLE IF NOT EXISTS insured_persons ( --被保险人唯一标识 guid TEXT PRIMARY KEY, --被保险人姓名 insured_person TEXT NOT NULL, --证件类型 identity_type TEXT NOT NULL, --证件号码 identity_number TEXT NOT NULL, --与主被保险人关系，包括本人、父母、配偶和子女等 relationship TEXT NOT NULL, --个单唯一标识，用于联查个单 person_policy_guid TEXT NOT NULL ) """ ) # 初始化责任表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS liabilities ( --责任唯一标识 guid TEXT PRIMARY KEY, --责任名称 liability TEXT NOT NULL, --出险事故 accident TEXT NOT NULL, --个人自费理算比例 personal_self_ratio TEXT NOT NULL, --个人自付理算比例 non_medical_ratio TEXT NOT NULL, --合理理算比例 reasonable_ratio TEXT NOT NULL, --理算保单唯一标识 adjust_policy_guid TEXT NOT NULL, --个单唯一标识 person_policy_guid TEXT NOT NULL ) """ ) # 初始化保额变动表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS coverage_changes ( --保额变动唯一标识 guid TEXT PRIMARY KEY, --变动类型，包括承保和理算等 change_type TEXT NOT NULL, --变动前金额 before_change_amount TEXT NOT NULL, --变动金额 change_amount TEXT NOT NULL, --变动后金额 after_change_amount TEXT NOT NULL, --变动时间 change_time TEXT NOT NULL, --变动保单唯一标识 change_policy_guid TEXT NOT NULL ) """ ) # 初始化购药及就医机构表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS institutions ( --购药及就医机构 institution TEXT PRIMARY KEY, --购药及就医机构类型 institution_type TEXT NOT NULL, --所在省 province TEXT NOT NULL, --所在市 city TEXT NOT NULL ) """ ) # 初始化药品表 self._execute( sql=""" CREATE TABLE IF NOT EXISTS medicines ( --药品/医疗服务 medicine TEXT PRIMARY KEY ) """ ) except Exception as exception: raise RuntimeError(f"初始化数据库发生异常：{str(exception)}") # noinspection PyShadowingNames def query_liabilities( self, insurer_company: str, insured_person: str, identity_type: str, identity_number: str, report_date: str, ) -> Optional[List[Dict[str, Any]]]: """ 根据保险分公司名称、被保险人姓名、证件类型、证件号码和出险时间查询个单和责任数据 :param insurer_company: 保险分公司名称 :param insured_person: 被保险人姓名 :param identity_type: 证件类型 :param identity_number: 证件号码 :param report_date: 报案时间 :return: 个单和责任数据 """ # noinspection PyBroadException try: with self: # noinspection SqlResolve results = self._query_all( sql=""" SELECT group_policies.group_policy, group_policies.insurer_company, person_policies.person_policy, person_policy_coverage_changes.after_change_amount AS remaining_amount, master_insured_persons.insured_person AS master_insured_person, insured_persons.insured_person, insured_persons.identity_type, insured_persons.identity_number, insured_persons.relationship, MAX(group_policies.commencement_date, person_policies.commencement_date) AS commencement_date, MIN(group_policies.termination_date, person_policies.termination_date) AS termination_date, person_policies.guid AS person_policy_guid, liabilities.liability, liabilities.accident, liabilities.personal_self_ratio, liabilities.non_medical_ratio, liabilities.reasonable_ratio, liabilities.adjust_policy_guid FROM insured_persons INNER JOIN insured_persons master_insured_persons ON person_policies.guid = master_insured_persons.person_policy_guid AND master_insured_persons.relationship = "本人" INNER JOIN person_policies ON insured_persons.person_policy_guid = person_policies.guid INNER JOIN group_policies ON person_policies.group_policy_guid = group_policies.guid INNER JOIN liabilities ON person_policies.guid = liabilities.person_policy_guid INNER JOIN coverage_changes person_policy_coverage_changes ON person_policies.guid = person_policy_coverage_changes.change_policy_guid AND person_policy_coverage_changes.change_time = (SELECT MAX(change_time) FROM coverage_changes WHERE change_policy_guid = person_policies.guid) INNER JOIN coverage_changes ON liabilities.adjust_policy_guid = coverage_changes.change_policy_guid AND coverage_changes.change_time = (SELECT MAX(change_time) FROM coverage_changes WHERE liabilities.adjust_policy_guid = change_policy_guid) WHERE group_policies.insurer_company = ? AND insured_persons.insured_person = ? AND insured_persons.identity_type = ? AND insured_persons.identity_number = ? AND ? BETWEEN group_policies.commencement_date AND group_policies.termination_date AND ? BETWEEN person_policies.commencement_date AND person_policies.termination_date AND CAST(coverage_changes.after_change_amount AS REAL) > 0 """, parameters=( insurer_company, insured_person, identity_type, identity_number, report_date, report_date, ), ) if results: # 就个人自费比例、个人自付比例和合理比例转为小数（decimal对象），保险起期、止期则转为日期时间（datetime对象） results = [ { k: ( Decimal(v).quantize( Decimal("0.00"), rounding=ROUND_HALF_UP, ) if k in [ "remaining_amount", "personal_self_ratio", "non_medical_ratio", "reasonable_ratio", ] else ( datetime.strptime(v, "%Y-%m-%d") if k in ["commencement_date", "termination_date"] else v ) ) for k, v in e.items() } for e in results ] person_policies = {} # 重构数据结构 for result in results: liabilities = { "liability": result["liability"], "accident": result["accident"], "personal_self_ratio": result["personal_self_ratio"], "non_medical_ratio": result["non_medical_ratio"], "reasonable_ratio": result["reasonable_ratio"], "adjust_policy_guid": result["adjust_policy_guid"], } person_policy_guid = result["person_policy_guid"] if person_policy_guid in person_policies: person_policies[person_policy_guid]["liabilities"].append( liabilities ) else: person_policies[person_policy_guid] = { "group_policy": result["group_policy"], "insurer_company": result["insurer_company"], "person_policy": result["person_policy"], "master_insured_person": result["master_insured_person"], "insured_person": result["insured_person"], "identity_type": result["identity_type"], "identity_number": result["identity_number"], "relationship": result["relationship"], "commencement_date": result["commencement_date"], "termination_date": result["termination_date"], "remaining_amount": result["remaining_amount"], "liabilities": [liabilities], } return [v for k, v in person_policies.items()] raise RuntimeError("查无数据") # TODO: 若根据保险分公司名称、被保险人姓名、证件类型、证件号码和出险时间查询被保险人发生异常则流转至主数据人工处理 except Exception as exception: raise RuntimeError(f"{str(exception)}") # noinspection PyShadowingNames def query_institution_type(self, institution: str) -> Optional[str]: """ 根据购药及就医机构查询购药及就医机构类型 :param institution: 购药及就医机构 :return: 购药及就医机构类型 """ # noinspection PyBroadException try: with self: # noinspection SqlResolve result = self._query_one( sql=""" SELECT institution_type FROM institutions WHERE institution = ? """, parameters=(institution,), ) if result: return result["institution_type"] raise # TODO: 若根据购药及就医机构查询购药及就医机构类型发生异常则流转至主数据人工处理 except Exception: raise # noinspection PyShadowingNames def query_medicine( self, content: str, ) -> Optional[str]: """ 根据明细项中具体内容查询药品/医疗服务 :param content: 明细项具体内容 :return: 药品/医疗服务 """ # TODO: 暂仅支持查询药品、通过药品/医疗服务包含明细项中具体内容查询 # noinspection PyBroadException try: with self: # noinspection SqlResolve result = self._query_all( sql=""" SELECT medicine FROM medicines WHERE ? LIKE '%' || medicine || '%' """, parameters=(content,), ) if result: return max(result, key=lambda x: len(x["medicine"]))[ "medicine" ] # 返回最大长度的药品/医疗服务 raise # TODO: 若根据明细项中具体内容查询药品/医疗服务发生异常则流转至主数据人工处理 except Exception: raise