Spaces:
Sleeping
Sleeping
File size: 4,686 Bytes
1462240 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
"""
Seed synonyms for search query expansion.
"""
import argparse
import csv
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Iterable, List, Tuple
import django
ROOT_DIR = Path(__file__).resolve().parents[2]
BACKEND_DIR = ROOT_DIR / "backend"
DATA_DIR = ROOT_DIR / "tài nguyên"
LOG_DIR = BACKEND_DIR / "logs" / "data_quality"
HUE_PORTAL_DIR = BACKEND_DIR / "hue_portal"
for path in (HUE_PORTAL_DIR, BACKEND_DIR, ROOT_DIR):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "hue_portal.hue_portal.settings")
django.setup()
from hue_portal.core.models import Synonym
DEFAULT_SEEDS: List[Tuple[str, str]] = [
("đèn đỏ", "vượt đèn đỏ"),
("vượt đèn", "vượt đèn đỏ"),
("nồng độ cồn", "rượu bia"),
("nồng độ cồn", "say xỉn"),
("nồng độ cồn", "uống rượu"),
("mũ bảo hiểm", "nón bảo hiểm"),
("mũ bảo hiểm", "mũ"),
("giấy phép lái xe", "bằng lái"),
("giấy phép lái xe", "GPLX"),
("giấy phép lái xe", "bằng"),
("đăng ký xe", "đăng ký"),
("đăng ký xe", "giấy đăng ký"),
("dừng đỗ", "đỗ xe"),
("dừng đỗ", "dừng xe"),
("dây an toàn", "thắt dây an toàn"),
("tốc độ", "vượt tốc độ"),
("tốc độ", "quá tốc độ"),
("sai làn", "sai đường"),
("sai làn", "đi sai làn"),
("điện thoại", "sử dụng điện thoại"),
("điện thoại", "gọi điện"),
("cư trú", "thủ tục cư trú"),
("cư trú", "đăng ký cư trú"),
("cư trú", "tạm trú"),
("cư trú", "thường trú"),
("ANTT", "an ninh trật tự"),
("ANTT", "an ninh"),
("PCCC", "phòng cháy chữa cháy"),
("PCCC", "cháy nổ"),
("thủ tục", "hành chính"),
("thủ tục", "TTHC"),
("công an", "CA"),
("công an", "cảnh sát"),
("tiếp dân", "tiếp công dân"),
("tiếp dân", "một cửa"),
("đơn vị", "cơ quan"),
("đơn vị", "phòng ban"),
]
def load_from_csv(path: Path) -> List[Tuple[str, str]]:
if not path.exists():
return []
pairs: List[Tuple[str, str]] = []
with path.open(encoding="utf-8") as handle:
reader = csv.DictReader(handle)
for row in reader:
keyword = (row.get("keyword") or "").strip()
alias = (row.get("alias") or "").strip()
if keyword and alias:
pairs.append((keyword, alias))
return pairs
def seed_synonyms(pairs: Iterable[Tuple[str, str]], log_path: Path) -> None:
created = 0
updated = 0
skipped = 0
with log_path.open("a", encoding="utf-8") as log_file:
for keyword, alias in pairs:
try:
synonym, was_created = Synonym.objects.get_or_create(
keyword=keyword,
defaults={"alias": alias}
)
if was_created:
created += 1
log_file.write(f"{datetime.utcnow().isoformat()}Z CREATED {keyword} -> {alias}\n")
else:
if synonym.alias != alias:
synonym.alias = alias
synonym.save(update_fields=["alias"])
updated += 1
log_file.write(f"{datetime.utcnow().isoformat()}Z UPDATED {keyword} -> {alias}\n")
else:
skipped += 1
except Exception as exc:
log_file.write(f"{datetime.utcnow().isoformat()}Z ERROR {keyword} -> {alias} :: {exc}\n")
total = Synonym.objects.count()
print(f"✅ Seeded {created} mới, cập nhật {updated}, bỏ qua {skipped}. Tổng: {total}")
print(f"Log chi tiết: {log_path}")
def parse_args():
parser = argparse.ArgumentParser(description="Seed synonyms cho chatbot")
parser.add_argument("--source", type=Path, default=DATA_DIR / "synonyms.csv", help="Đường dẫn CSV synonyms")
parser.add_argument("--include-default", action="store_true", help="Bao gồm seed mặc định trong script")
return parser.parse_args()
def main():
args = parse_args()
LOG_DIR.mkdir(parents=True, exist_ok=True)
log_path = LOG_DIR / f"synonyms_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.log"
pairs: List[Tuple[str, str]] = []
csv_pairs = load_from_csv(args.source)
if csv_pairs:
pairs.extend(csv_pairs)
if args.include_default or not pairs:
pairs.extend(DEFAULT_SEEDS)
seed_synonyms(pairs, log_path)
if __name__ == "__main__":
main()
|