zk
92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
"""字典数据缓存服务
|
|
|
|
启动时从 MySQL 加载岗位分类、行业、专业分类、地区数据到内存。
|
|
"""
|
|
|
|
from sqlalchemy import select, text
|
|
|
|
from app.core.database import MysqlSession
|
|
from app.core.logger import log
|
|
|
|
|
|
class DictCacheService:
|
|
"""字典缓存，单例使用"""
|
|
|
|
def __init__(self):
|
|
self.job_category_text: str = ""
|
|
self.industry_text: str = ""
|
|
self.major_category_text: str = ""
|
|
self._region_list: list[dict] = []
|
|
|
|
async def refresh(self) -> None:
|
|
"""加载全量字典数据"""
|
|
async with MysqlSession() as session:
|
|
# 岗位分类（三级叶子，带父级路径）
|
|
result = await session.execute(text("""
|
|
SELECT c.id, c.name, c.parent_id, c.root_id, c.level,
|
|
p.name AS parent_name, r.name AS root_name
|
|
FROM bg_job_category c
|
|
LEFT JOIN bg_job_category p ON c.parent_id = p.id
|
|
LEFT JOIN bg_job_category r ON c.root_id = r.id
|
|
WHERE c.level = 3
|
|
"""))
|
|
categories = result.mappings().all()
|
|
self.job_category_text = ", ".join(
|
|
f"{c['id']}:{c['name']}({c['root_name']}/{c['parent_name']})"
|
|
for c in categories
|
|
)
|
|
|
|
# 行业（二级叶子，带父级）
|
|
result = await session.execute(text("""
|
|
SELECT i.id, i.name, p.name AS parent_name
|
|
FROM bg_industry i
|
|
LEFT JOIN bg_industry p ON i.parent_id = p.id
|
|
WHERE i.level = 2
|
|
"""))
|
|
industries = result.mappings().all()
|
|
self.industry_text = ", ".join(
|
|
f"{i['id']}:{i['name']}({i['parent_name']})"
|
|
for i in industries
|
|
)
|
|
|
|
# 专业分类（三级叶子，带父级路径）
|
|
result = await session.execute(text("""
|
|
SELECT m.id, m.name, m.parent_id, m.root_id,
|
|
p.name AS parent_name, r.name AS root_name
|
|
FROM bg_major_category m
|
|
LEFT JOIN bg_major_category p ON m.parent_id = p.id
|
|
LEFT JOIN bg_major_category r ON m.root_id = r.id
|
|
WHERE m.level = 3
|
|
"""))
|
|
majors = result.mappings().all()
|
|
self.major_category_text = ", ".join(
|
|
f"{m['id']}:{m['name']}({m['root_name']}/{m['parent_name']})"
|
|
for m in majors
|
|
)
|
|
|
|
# 地区（省市级）
|
|
result = await session.execute(text("""
|
|
SELECT code, name FROM bg_china_regions_code WHERE city_code IS NULL
|
|
"""))
|
|
self._region_list = [dict(r) for r in result.mappings().all()]
|
|
|
|
log.info(
|
|
"字典缓存加载完成: 岗位分类{}条, 行业{}条, 专业{}条, 地区{}条",
|
|
len(categories), len(industries), len(majors), len(self._region_list),
|
|
)
|
|
|
|
def match_region_code(self, city_name: str) -> str | None:
|
|
"""根据城市名模糊匹配地区编码"""
|
|
if not city_name:
|
|
return None
|
|
name = city_name.replace("市", "").replace("省", "").strip()
|
|
for r in self._region_list:
|
|
r_name = r["name"].replace("市", "").replace("省", "")
|
|
if name in r_name or r_name in name:
|
|
return r["code"]
|
|
return None
|
|
|
|
|
|
# 全局单例
|
|
dict_cache = DictCacheService()
|