import os import re from pathlib import Path import pandas as pd from utils.log import log YEAR_PATTERN = re.compile(r"^\d{4}$") MONTH_PATTERN = re.compile(r"^(0[1-9]|1[0-2])$") def format_sql_value(value): """ 将 Python 值转换为 SQL 可识别的格式： - None -> NULL - 数值 -> 原样输出 - 字符串 -> 加单引号 """ if pd.isna(value): return 'NULL' elif isinstance(value, (int, float)): return str(value) else: return f"'{value}'" def get_previous_month_dir(current_path): """生成前月目录路径""" try: year_part = current_path.parent.name month_part = current_path.name if not (YEAR_PATTERN.match(year_part) and MONTH_PATTERN.match(month_part)): return None prev_month = int(month_part) - 1 if prev_month < 1: return None return current_path.parent.parent / current_path.parent.name / f"{prev_month:02d}" except Exception as e: log.info(f"前月目录生成失败：{str(e)}") return None COUNTRY_CODE_MAPPING = { # ================= 亚洲 ================= "阿富汗": "AF", "巴林": "BH", "孟加拉国": "BD", "不丹": "BT", "文莱": "BN", "缅甸": "MM", "柬埔寨": "KH", "塞浦路斯": "CY", "朝鲜": "KP", "中国香港": "HK", "印度": "IN", "印度尼西亚": "ID", "伊朗": "IR", "伊拉克": "IQ", "以色列": "IL", "日本": "JP", "约旦": "JO", "科威特": "KW", "老挝": "LA", "黎巴嫩": "LB", "中国澳门": "MO", "马来西亚": "MY", "马尔代夫": "MV", "蒙古": "MN", "尼泊尔": "NP", "阿曼": "OM", "巴基斯坦": "PK", "巴勒斯坦": "PS", "菲律宾": "PH", "卡塔尔": "QA", "沙特阿拉伯": "SA", "新加坡": "SG", "韩国": "KR", "斯里兰卡": "LK", "叙利亚": "SY", "泰国": "TH", "土耳其": "TR", "阿联酋": "AE", "也门": "YE", "越南": "VN", "中国": "CN", "中国台湾": "TW", "哈萨克斯坦": "KZ", "吉尔吉斯斯坦": "KG", "塔吉克斯坦": "TJ", "土库曼斯坦": "TM", "乌兹别克斯坦": "UZ", "格鲁吉亚": "GE", "亚美尼亚": "AM", "阿塞拜疆": "AZ", # ================= 非洲 ================= "阿尔及利亚": "DZ", "安哥拉": "AO", "贝宁": "BJ", "博茨瓦纳": "BW", "布隆迪": "BI", "喀麦隆": "CM", "佛得角": "CV", "中非": "CF", "乍得": "TD", "科摩罗": "KM", "刚果共和国": "CG", "吉布提": "DJ", "埃及": "EG", "赤道几内亚": "GQ", "埃塞俄比亚": "ET", "加蓬": "GA", "冈比亚": "GM", "加纳": "GH", "几内亚": "GN", "几内亚比绍": "GW", "科特迪瓦": "CI", "肯尼亚": "KE", "莱索托": "LS", "利比里亚": "LR", "利比亚": "LY", "马达加斯加": "MG", "马拉维": "MW", "马里": "ML", "毛里塔尼亚": "MR", "毛里求斯": "MU", "摩洛哥": "MA", "莫桑比克": "MZ", "纳米比亚": "NA", "尼日尔": "NE", "尼日利亚": "NG", "卢旺达": "RW", "圣多美和普林西比": "ST", "塞内加尔": "SN", "塞舌尔": "SC", "塞拉利昂": "SL", "索马里": "SO", "南非": "ZA", "苏丹": "SD", "坦桑尼亚": "TZ", "多哥": "TG", "突尼斯": "TN", "乌干达": "UG", "布基纳法索": "BF", "刚果民主共和国": "CD", "赞比亚": "ZM", "津巴布韦": "ZW", "厄立特里亚": "ER", "南苏丹": "SS", # ================= 欧洲 ================= "比利时": "BE", "丹麦": "DK", "英国": "GB", "德国": "DE", "法国": "FR", "爱尔兰": "IE", "意大利": "IT", "卢森堡": "LU", "荷兰": "NL", "希腊": "GR", "葡萄牙": "PT", "西班牙": "ES", "阿尔巴尼亚": "AL", "奥地利": "AT", "保加利亚": "BG", "芬兰": "FI", "匈牙利": "HU", "冰岛": "IS", "列支敦士登": "LI", "马耳他": "MT", "挪威": "NO", "波兰": "PL", "罗马尼亚": "RO", "瑞典": "SE", "瑞士": "CH", "爱沙尼亚": "EE", "拉脱维亚": "LV", "立陶宛": "LT", "白俄罗斯": "BY", "摩尔多瓦": "MD", "俄罗斯": "RU", "乌克兰": "UA", "斯洛文尼亚": "SI", "克罗地亚": "HR", "捷克": "CZ", "斯洛伐克": "SK", "北马其顿": "MK", "波斯尼亚和黑塞哥维那": "BA", "梵蒂冈": "VA", "塞尔维亚": "RS", "黑山": "ME", # ================= 美洲 ================= "安提瓜和巴布达": "AG", "阿根廷": "AR", "巴哈马": "BS", "巴巴多斯": "BB", "伯利兹": "BZ", "玻利维亚": "BO", "巴西": "BR", "加拿大": "CA", "智利": "CL", "哥伦比亚": "CO", "哥斯达黎加": "CR", "古巴": "CU", "多米尼克": "DM", "多米尼加": "DO", "厄瓜多尔": "EC", "萨尔瓦多": "SV", "格林纳达": "GD", "危地马拉": "GT", "圭亚那": "GY", "海地": "HT", "洪都拉斯": "HN", "牙买加": "JM", "墨西哥": "MX", "尼加拉瓜": "NI", "巴拿马": "PA", "巴拉圭": "PY", "秘鲁": "PE", "圣卢西亚": "LC", "圣文森特和格林纳丁斯": "VC", "苏里南": "SR", "特立尼达和多巴哥": "TT", "美国": "US", "乌拉圭": "UY", "委内瑞拉": "VE", "圣基茨和尼维斯": "KN", # ================= 大洋洲 ================= "澳大利亚": "AU", "斐济": "FJ", "基里巴斯": "KI", "马绍尔群岛": "MH", "密克罗尼西亚联邦": "FM", "瑙鲁": "NR", "新西兰": "NZ", "帕劳": "PW", "巴布亚新几内亚": "PG", "萨摩亚": "WS", "所罗门群岛": "SB", "汤加": "TO", "图瓦卢": "TV", "瓦努阿图": "VU", # ================= 特殊地区 ================= "法属圭亚那": "GF", "瓜德罗普": "GP", "留尼汪": "RE", "圣马丁": "MF", "荷属圣马丁": "SX", "法属波利尼西亚": "PF", "新喀里多尼亚": "NC", "库克群岛": "CK", "关岛": "GU", "波多黎各": "PR", "美属萨摩亚": "AS", "百慕大": "BM", "开曼群岛": "KY", "福克兰群岛（马尔维纳斯）": "FK", "格陵兰": "GL", "法属南方领地": "TF", "赫德岛和麦克唐纳岛": "HM", "托克劳": "TK", "纽埃": "NU", "诺福克岛": "NF", "北马里亚纳群岛": "MP", "皮特凯恩": "PN", "圣赫勒拿": "SH", "斯瓦尔巴群岛和扬马延岛": "SJ", "东帝汶": "TL", # ==== 欧洲特殊地区 ==== "加那利群岛": "IC", # 西班牙特殊领土代码 "塞卜泰(休达)": "XC", # 休达官方代码 "梅利利亚": "XL", # 梅利利亚官方代码 "安道尔": "AD", "直布罗陀": "GI", "摩纳哥": "MC", "圣马力诺": "SM", "法罗群岛": "FO", # 丹麦自治领 "奥兰群岛": "AX", # 芬兰自治省 "格恩西": "GG", # 英国皇家属地 "马恩岛": "IM", "泽西": "JE", # ==== 非洲特殊地区 ==== "西撒哈拉": "EH", # 争议地区代码 "斯威士兰": "SZ", # 正式国名为"Eswatini"但保留旧映射 "马约特": "YT", # 法国海外省 # ==== 美洲特殊地区 ==== "英属印度洋领地": "IO", "阿鲁巴": "AW", "库拉索": "CW", "马提尼克": "MQ", # 法国海外省 "蒙特塞拉特": "MS", "法属圣马丁": "MF", "特克斯和凯科斯群岛": "TC", "英属维尔京群岛": "VG", "博纳尔，圣俄斯塔休斯和萨巴": "BQ", "圣巴泰勒米": "BL", # 法国海外集体 "美属维尔京群岛": "VI", "安圭拉": "AI", "圣皮埃尔和密克隆": "PM", # ==== 大洋洲特殊地区 ==== "瓦利斯和富图纳": "WF", "科科斯（基林）群岛": "CC", "圣诞岛": "CX", "美国本土外小岛屿": "UM", # ==== 特殊标记 ==== "布维岛": "BV", # 挪威属地 "南乔治亚岛和南桑德韦奇岛": "GS", "国家（地区）不明": "XX" # 自定义代码 } def extract_year_month_from_path(path): parts = path.parts try: year_part = parts[-2] month_part = parts[-1] if not YEAR_PATTERN.match(year_part): raise ValueError(f"无效年份格式：{year_part}") if not MONTH_PATTERN.match(month_part): raise ValueError(f"无效月份格式：{month_part}") return int(year_part), int(month_part) except IndexError: raise ValueError("路径结构不符合要求，示例：.../shandong/2025/04") download_dir = os.path.abspath(os.path.join('downloads')) download_dir_find = os.path.abspath(os.path.join('downloads/demo')) if __name__ == '__main__': year, month = extract_year_month_from_path(Path(download_dir)/'2025'/'02') log.info(year, month)