69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
import sys
|
|
import os
|
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
|
# Add backend to path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.models import SessionLocal, init_db
|
|
from app.routers.etl import import_data, ImportRequest
|
|
from app.services.excel_parser import excel_parser
|
|
|
|
def test_import():
|
|
db = SessionLocal()
|
|
try:
|
|
# 1. Upload/Parse DIT
|
|
base_dir = Path(__file__).resolve().parent.parent
|
|
dit_path = base_dir / "data" / "uploads" / "DIT report (by Focus Item)_app_樣本 2.xlsx"
|
|
print(f"Parsing {dit_path}...")
|
|
file_id, info = excel_parser.parse_file(dit_path, "dit")
|
|
print(f"File ID: {file_id}, Rows: {info['row_count']}")
|
|
|
|
# 2. Import DIT
|
|
print("Importing DIT...")
|
|
req = ImportRequest(file_id=file_id)
|
|
res = import_data(req, db)
|
|
print(f"Imported {res.imported_count} DIT records.")
|
|
|
|
# 3. Upload/Parse Sample
|
|
sample_path = base_dir / "data" / "uploads" / "樣品申請紀錄_樣本_9月.xlsx"
|
|
print(f"Parsing {sample_path}...")
|
|
file_id_s, info_s = excel_parser.parse_file(sample_path, "sample")
|
|
print(f"File ID: {file_id_s}, Rows: {info_s['row_count']}")
|
|
|
|
# 4. Import Sample
|
|
print("Importing Sample...")
|
|
req_s = ImportRequest(file_id=file_id_s)
|
|
res_s = import_data(req_s, db)
|
|
print(f"Imported {res_s.imported_count} Sample records.")
|
|
|
|
# 5. Upload/Parse Order
|
|
order_path = base_dir / "data" / "uploads" / "訂單樣本_20251217_調整.xlsx"
|
|
print(f"Parsing {order_path}...")
|
|
file_id_o, info_o = excel_parser.parse_file(order_path, "order")
|
|
print(f"File ID: {file_id_o}, Rows: {info_o['row_count']}")
|
|
|
|
# 6. Import Order
|
|
print("Importing Order...")
|
|
req_o = ImportRequest(file_id=file_id_o)
|
|
res_o = import_data(req_o, db)
|
|
print(f"Imported {res_o.imported_count} Order records.")
|
|
|
|
# 7. Run Matching
|
|
from app.services.fuzzy_matcher import FuzzyMatcher
|
|
print("Running Matching...")
|
|
matcher = FuzzyMatcher(db)
|
|
match_res = matcher.run_matching()
|
|
print(f"Matching completed: {match_res}")
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
print("Error during test:")
|
|
print(traceback.format_exc())
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
test_import()
|