# -*- coding: utf-8 -*- """属性处理纯函数（PURE FUNCTIONS）。 按 ARCHITECTURE.md Phase 6 拆分建议，从 ``interface/getAttributes.py`` 迁出 被 ``predictors/`` 反向导入的纯函数，消除 ``predictors ↔ interface.getAttributes`` 循环依赖。 原位置：``interface/getAttributes.py``： - ``turnMoneySource`` — 资金来源文本归一（原 3335-3368 行） - ``get_days_between`` — 计算两日期天数差（原 4211-4228 行） - ``extract_serviceTime`` — 解析服务期（起止日期/天数）（原 4230-4407 行） 迁移原因： - ``predictors/table_prem.py`` 导入 ``extract_serviceTime`` - ``predictors/approval.py`` 导入 ``turnMoneySource`` - ``template_extractor.py`` 导入 ``extract_serviceTime`` - ``interface/predictor.py`` 导入 ``turnMoneySource``, ``extract_serviceTime`` 将这些函数下沉到 ``common/`` 层，使 ``predictors/`` 只依赖 ``common/``， 不再反向依赖 ``interface.getAttributes``（属于 assembly 层）。 按 ARCHITECTURE.md §4.3 依赖方向约束： common 可被各层引用，但不反向依赖 pipeline / predictors / assembly / rules。 ``interface/getAttributes.py`` 仍 re-export 以上全部名称，老 import 不受影响。 """ from __future__ import absolute_import import re import calendar from datetime import datetime from BiddingKG.dl.common.Utils import isValidDate __all__ = [ "turnMoneySource", "get_days_between", "extract_serviceTime", ] def turnMoneySource(moneysource): result_list = [] if re.search("自筹|业主筹集|筹资|自有", moneysource): result_list.append("自筹") if re.search("财政", moneysource) and not re.search("非财政", moneysource): result_list.append("财政资金") if re.search("拨款|补助|划拨|拨付|国拨|上级资金", moneysource): result_list.append("上级拨款") if re.search("社会资本|社会资金", moneysource): result_list.append("社会资本") if re.search("贷款|借款|借贷", moneysource): result_list.append("贷款资金") if re.search("债券|债|国债", moneysource): result_list.append("债券资金") if re.search("专项|项目资金", moneysource): result_list.append("项目专项资金") if re.search("配套", moneysource): result_list.append("配套资金") if re.search("外资", moneysource): result_list.append("外资") if re.search("国有资金|国企资金|国资|国家投资", moneysource): result_list.append("国有资金") if re.search("投资|融资", moneysource): result_list.append("投资资金") if re.search("预算(? 0 and len(result_list) < 5: return ",".join(result_list) else: return "其他资金" def get_days_between(day1, day2, get_abs=0): ''' :param day1: 较小日期 :param day2: 较大日期 :param get_abs: 是否取绝对值 :return: 天数差 ''' # 将日期字符串转换为datetime对象 date1 = datetime.strptime(day1, '%Y-%m-%d') date2 = datetime.strptime(day2, '%Y-%m-%d') # 计算日期差 delta = date2 - date1 # 获取天数差 days_difference = delta.days if get_abs: return abs(days_difference) else: return days_difference def extract_serviceTime(service_time, page_time): pattern1 = re.compile("\d{4}[年\-./]\d{1,2}[月\-./]\d{1,2}日?") pattern2 = re.compile("\d+(?:\.\d+)?[(（]?个?[^\d]?[^\d]?(?:日|天|周年|整年|学?年|月|周|日历[天日]|工作[天日])") pattern3 = re.compile("\d{4}[年\-./]\d{1,2}月?") pattern4 = re.compile("(?:日|天|周年|年|月|周|日历[天日]|工作[天日]|星期)[^\d]{1,3}\d+(?:\.\d+)?") DigitsDic = {"零": 0, "壹": 1, "贰": 2, "叁": 3, "肆": 4, "伍": 5, "陆": 6, "柒": 7, "捌": 8, "玖": 9, "〇": 0, "一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "两": 2, '貮': 2} def get_month_days(year, month): # calendar.monthrange(year, month)返回一个元组，其中第一个元素是月份的第一天是星期几（0-6为星期一到星期日）， # 第二个元素是该月的天数。 _, last_day = calendar.monthrange(year, month) return last_day def get_num(text): CN_UNIT = {'十': 10, '拾': 10, '百': 100, '佰': 100, '千': 1000, '仟': 1000} regex = re.compile(r'[〇一二三四五六七八九零壹贰叁肆伍陆柒捌玖貮两十拾百佰千仟]+') text = regex.search(text) if text: text = text.group() else: return "" result = 0 result_list = [] unit = 0 control = 0 for i, d in enumerate(text): if d in '零百佰千仟' and i == 0: return "" if d in DigitsDic: result += DigitsDic[d] elif d in CN_UNIT: if unit == 0: unit_1 = CN_UNIT[d] # 这里的处理主要是考虑到类似于二十三亿五千万这种数 if result == 0: result = CN_UNIT[d] else: result *= CN_UNIT[d] unit = CN_UNIT[d] result_1 = result elif unit > CN_UNIT[d]: result -= DigitsDic[text[i - 1]] result += DigitsDic[text[i - 1]] * CN_UNIT[d] unit = CN_UNIT[d] elif unit <= CN_UNIT[d]: if (CN_UNIT[d] < unit_1) and (len(result_list) == control): result_list.append(result_1) result = (result - result_1) * CN_UNIT[d] control += 1 else: result *= CN_UNIT[d] unit = CN_UNIT[d] if len(result_list) == control: unit_1 = unit result_1 = result else: return "" return sum(result_list) + result serviceTime_dict = {"service_start": "", "service_end": "", "service_days": 0} re_num = re.findall(r'[〇一二三四五六七八九零壹贰叁肆伍陆柒捌玖貮两十拾百佰千仟]+', service_time) for _num in re_num: if not re.search("[十拾百佰千仟]", _num): num = "" for word in _num: num += str(DigitsDic.get(word, word)) service_time = service_time.replace(_num, num, 1) else: num = str(get_num(_num)) service_time = service_time.replace(_num, num, 1) end_time = "" service_days = 0 re_page_time = re.search("20\d{2}-\d{2}-\d{2}", page_time) page_time = re_page_time.group() if re_page_time else "2000-01-01" # page_time为空时默认值为2000-01-01 if re.search(pattern1, service_time): # end_time = re.findall(pattern1,service_time)[-1] time_list = [] for _time in re.findall(pattern1, service_time): _time = re.sub("日", "", _time) _time = re.sub("[年月./]", "-", _time) _year, _month, _day = _time.split("-") _month = int(_month) _day = int(_day) _year = int(_year) if _year > 2050 or _year <= 2000 or _month > 12 or _month <= 0 or _day <= 0 or _day > 31: service_days = 0 else: if isValidDate(_year, _month, _day): _time = str(_year) + '-' + str(_month) + '-' + str(_day) _time = _time.split("-")[0] + '-' + _time.split("-")[1].rjust(2, "0") + '-' + _time.split("-")[ 2].rjust(2, "0") time_list.append(_time) if len(time_list) >= 2: if get_days_between(page_time, time_list[1]) > 1 and get_days_between(time_list[0], time_list[1]) > 0: serviceTime_dict['service_end'] = time_list[1] serviceTime_dict['service_start'] = time_list[0] elif len(time_list) == 1: if get_days_between(page_time, time_list[0]) > 1: serviceTime_dict['service_end'] = time_list[0] # service_days = (time.mktime(time.strptime(end_time,"%Y-%m-%d"))-page_timestamp)/(24*60*60) elif re.search(pattern3, service_time): time_list = [] # end_time = re.findall(pattern3,service_time)[-1] for _time in re.findall(pattern3, service_time): _time = re.sub("月", "", _time) _time = re.sub("[年./]", "-", _time) _year, _month = _time.split("-") _day = 0 _month = int(_month) _year = int(_year) if _year > 2050 or _year <= 2000 or _month > 12 or _month <= 0: service_days = 0 else: _day = get_month_days(_year, _month) if isValidDate(_year, _month, _day): _time = str(_year) + '-' + str(_month) + '-' + str(_day) _time = _time.split("-")[0] + '-' + _time.split("-")[1].rjust(2, "0") + '-' + _time.split("-")[ 2].rjust(2, "0") time_list.append(_time) if len(time_list) >= 2: if get_days_between(page_time, time_list[1]) > 1 and get_days_between(time_list[0], time_list[1]) > 0: serviceTime_dict['service_end'] = time_list[1] serviceTime_dict['service_start'] = time_list[0] elif len(time_list) == 1: if get_days_between(page_time, time_list[0]) > 1: serviceTime_dict['service_end'] = time_list[0] # service_days = (time.mktime(time.strptime(end_time,"%Y-%m-%d"))-page_timestamp)/(24*60*60) elif re.search(pattern2, service_time) or re.search(pattern4, service_time): for pattern in [pattern2, pattern4]: unit = 1 match = re.findall(pattern, service_time) if len(set(match)) == 1: match_text = match[0] if "月" in match_text: unit = 30 elif "年" in match_text: unit = 365 elif "周" in match_text or "星期" in match_text: unit = 7 match_num = float(re.search("\d+", match_text).group()) # 数字能被365整除，单位更正为天 if int(match_num) % 365 == 0: unit = 1 if unit == 365: if match_num > 10: # 单位为'年'时，排除数字过大的 match_num = 0 elif unit == 30: if match_num > 60: # 单位为'月'时，排除数字过大的 match_num = 0 elif unit == 1: if match_num > 4000: # 单位为'日'时，排除数字过大的 match_num = 0 service_days = int(match_num * unit) if service_days % 360 == 0: service_days = service_days / 360 * 365 elif service_days % 180 == 0 and service_days % 360 != 0: service_days = service_days // 360 * 365 + 180 service_days = int(service_days) if service_days <= 1 and service_days > 4000: service_days = 0 # if service_days>3: if service_days > 0: # service_days = str(service_days) + "天" serviceTime_dict['service_days'] = service_days break elif "半年" in service_time: service_days = 180 # service_days = str(service_days) + "天" serviceTime_dict['service_days'] = service_days if serviceTime_dict['service_start'] and serviceTime_dict['service_end']: service_days = get_days_between(serviceTime_dict['service_start'], serviceTime_dict['service_end']) serviceTime_dict['service_days'] = service_days return serviceTime_dict