Intention/api/globalData.py

# globalData.py
import json
import logging

import redis

from logger_util import setup_logger
from config import redis_url

logger = setup_logger("GlobalData", level=logging.DEBUG)


class GlobalData:
    # 数据字段
    standard_company_program = {}
    standard_company_name_list = []
    simply_to_standard_company_name_map = {}
    pinyin_simply_to_standard_company_name_map = {}

    standard_project_name_list = []
    simply_to_standard_project_name_map = {}
    pinyin_simply_to_standard_project_name_map = {}

    standard_construct_name_list = []
    simply_to_standard_construct_name_map = {}
    pinyin_simply_to_standard_construct_name_map = {}

    standard_constractor_name_list = []
    simply_to_standard_constractor_name_map = {}
    pinyin_simply_to_standard_constractor_name_map = {}

    standard_team_leader_name_list = []
    simply_to_standard_team_leader_name_map = {}
    pinyin_simply_to_standard_team_leader_name_map = {}

    standard_design_pic = {}

    @classmethod
    def update_from_redis(cls):
        import sys
        from utils import (
            clean_useless_company_name,
            clean_useless_project_name,
            clean_useless_team_leader_name
        )
        #实施组织（分公司）和项目名
        cls._update_company_program()
        #工程名
        cls._update_list_data('SBD_QUERY_DATA:PROJECT_NAME', './standard_data/standard_project.txt',
                              cls.standard_project_name_list, cls.simply_to_standard_project_name_map,
                              cls.pinyin_simply_to_standard_project_name_map, clean_useless_project_name)
        logger.info(f"工程名数量：{len(cls.standard_project_name_list)}")

        #建管单位
        cls._update_list_data('SBD_QUERY_DATA:CONSTRUCTION_UNIT', './standard_data/construct_unit.txt',
                              cls.standard_construct_name_list, cls.simply_to_standard_construct_name_map,
                              cls.pinyin_simply_to_standard_construct_name_map, clean_useless_company_name)
        logger.info(f"建管单位数量：{len(cls.standard_construct_name_list)}")

        #分包单位
        cls._update_list_data('SBD_QUERY_DATA:SUBCONTRACTOR', './standard_data/sub_contract.txt',
                              cls.standard_constractor_name_list, cls.simply_to_standard_constractor_name_map,
                              cls.pinyin_simply_to_standard_constractor_name_map, clean_useless_company_name)
        logger.info(f"分包单位数量：{len(cls.standard_constractor_name_list)}")

        #班组名称
        cls._update_list_data('SBD_QUERY_DATA:TEAM', './standard_data/team_leader.txt',
                              cls.standard_team_leader_name_list, cls.simply_to_standard_team_leader_name_map,
                              cls.pinyin_simply_to_standard_team_leader_name_map, clean_useless_team_leader_name)

        cls.update_design_pic_info()
        logger.info(f"班组名称数量：{len(cls.standard_team_leader_name_list)}")

    @classmethod
    def _update_company_program(cls):
        from utils import (
            load_standard_json_data,
            save_dict_to_file,
            clean_useless_company_name,
            text_to_pinyin
        )
        # 公司与项目关系数据
        try:
            r = redis.from_url(redis_url, decode_responses=True)
            json_str = r.get('SBD_QUERY_DATA:STANDARD_COMPANY_PROGRAM')
            if json_str:
                temp_data = json.loads(json_str)
                save_dict_to_file(temp_data, "./standard_data/standard_company_program.json")
                logger.info("[Info] Loaded STANDARD_COMPANY_PROGRAM from Redis")
            else:
                raise ValueError("Redis key not found")
        except Exception as e:
            logger.error(f"[Error] Error loading STANDARD_COMPANY_PROGRAM: {e}")
            temp_data = load_standard_json_data("./standard_data/standard_company_program.json")

        if temp_data != cls.standard_company_program:
            cls.standard_company_program.clear()
            cls.standard_company_program.update(temp_data)

            cls.standard_company_name_list.clear()
            cls.standard_company_name_list.extend(list(temp_data.keys()))

            cls.simply_to_standard_company_name_map.clear()
            cls.simply_to_standard_company_name_map.update({
                clean_useless_company_name(kw): kw for kw in cls.standard_company_name_list
            })

            cls.pinyin_simply_to_standard_company_name_map.clear()
            cls.pinyin_simply_to_standard_company_name_map.update({
                text_to_pinyin(clean_useless_company_name(kw)): kw for kw in cls.standard_company_name_list
            })

    @classmethod
    def _update_list_data(cls, redis_key, local_path, target_list, simple_map, pinyin_map, cleaner):
        from utils import (
            load_standard_name_list,
            save_standard_name_list_to_file,
            text_to_pinyin
        )
        try:
            r = redis.from_url(redis_url, decode_responses=True)
            json_str = r.get(redis_key)
        except Exception as e:
            logger.info(f"[Error] Redis error on key '{redis_key}': {e}")
            json_str = None

        if json_str:
            try:
                temp_list = json.loads(json_str)
                save_standard_name_list_to_file(temp_list, local_path)
                logger.info(f"[Info] Loaded {redis_key} from Redis")
            except json.JSONDecodeError as e:
                logger.info(f"[Warning] JSON decode error on key '{redis_key}': {e}")
                temp_list = load_standard_name_list(local_path)
        else:
            logger.info(f"[Info] Redis key '{redis_key}' not found. Loading from local file...")
            temp_list = load_standard_name_list(local_path)

        if temp_list != target_list:
            target_list.clear()
            target_list.extend(temp_list)

            simple_map.clear()
            simple_map.update({
                cleaner(kw): kw for kw in temp_list
            })

            pinyin_map.clear()
            pinyin_map.update({
                text_to_pinyin(cleaner(kw)): kw for kw in temp_list
            })

    @classmethod
    def update_design_pic_info(cls):

        from utils import (
            load_standard_json_data,
            save_dict_to_file,
            clean_useless_company_name,
            text_to_pinyin
        )
        # 公司与工程关系数据
        try:
            r = redis.from_url(redis_url, decode_responses=True)
            json_str = r.get('SBD_QUERY_DATA:STANDARD_DESIGN_PIC_INFO')
            if json_str:
                temp_data = json.loads(json_str)
                save_dict_to_file(temp_data, "./standard_data/standard_project_info.json")
                logger.info("[Info] Loaded STANDARD_DESIGN_PIC_INFO from Redis")
            else:
                raise ValueError("Redis key not found")
        except Exception as e:
            logger.error(f"[Error] Error loading STANDARD_DESIGN_PIC_INFO: {e}")
            temp_data = load_standard_json_data("./standard_data/standard_project_info.json")

        print(f"STANDARD_DESIGN_PIC_INFO:{temp_data}")
        if temp_data != cls.standard_design_pic:
            cls.standard_design_pic.clear()
            cls.standard_design_pic.update(temp_data)

    @classmethod
    def get_all_company_from_design_info(cls):
        #获取所有分公司信息
        return list(cls.standard_design_pic.keys())

    @classmethod
    def get_project_from_design_info(cls):
        #获取所有工程信息
        project_list = []

        for company, projects in cls.standard_design_pic.items():
            project_list.extend(projects.keys())
        return project_list


    @classmethod
    def get_contents_by_company_proj(cls, company_name, project_name):
        #根据分公司名和工程名 获取方案，图纸和规范规程
        global company_projects
        if company_name and project_name:
            return cls.standard_design_pic.get(company_name, {}).get(project_name, {})
        elif not company_name and project_name:
            for company, projects in cls.standard_design_pic.items():
                if project_name in projects:
                    return projects[project_name]  # 返回该工程下的“方案/图纸/规范规程”
        elif company_name and not project_name:
            result = {
                "方案": [],
                "图纸": [],
                "规范规程": []
            }
            company_projects = cls.standard_design_pic.get(company_name, {})
            # 忽略 "@type" 字段
            if "@type" in company_projects:
                del company_projects["@type"]

            for proj_key, project in company_projects.items():
                if "@type" in project:
                    del project["@type"]
                for key in result:
                    result[key].extend(project.get(key, []))

            # print(f"最终result：{result}")
            return result
        else:
            return None


    @classmethod
    def get_contents_by_proj(cls, project_name):
        #根据工程名 获取方案，图纸和规范规程
        for company, projects in cls.standard_design_pic.items():
            if project_name in projects:
                return projects[project_name]  # 返回该工程下的“方案/图纸/规范规程”
            return None  # 没找到