""" Server trung gian giua prototype va globalfastener.com Chay: python fastener_server.py Mo prototype tai: http://localhost:8899/prototype.html Chuc nang: - Serve file HTML/JS/CSS cua prototype - API /api/standards?sort=55 -> lay danh sach tieu chuan theo nhom - API /api/detail?sid=xxx -> lay thong so ky thuat 1 tieu chuan - Cache local: fetch 1 lan, lan sau lay tu cache """ import http.server import json import os import re import sys import time import hashlib from urllib.parse import urlparse, parse_qs, urljoin # Fix encoding Windows os.environ["PYTHONIOENCODING"] = "utf-8" if sys.stdout.encoding != "utf-8": sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") import requests from bs4 import BeautifulSoup PORT = 8899 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) CACHE_DIR = os.path.join(BASE_DIR, ".spec_cache") GF_BASE = "https://www.globalfastener.com" HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} # Tao thu muc cache os.makedirs(CACHE_DIR, exist_ok=True) # ===== Danh muc tieu chuan ===== STANDARD_GROUPS = { # === Bu lông === "hex_bolts": {"sort": 6, "name": "Hex Bolts (Bu lông lục giác)"}, "flange_bolts": {"sort": 121, "name": "Flanged Hex Bolts (Bu lông mặt bích)"}, "structural_bolts": {"sort": 124, "name": "Structural Bolts (Bu lông kết cấu)"}, "square_bolts": {"sort": 14, "name": "Square Head Bolts (Bu lông đầu vuông)"}, "eye_bolts": {"sort": 15, "name": "Eye Bolts (Bu lông mắt)"}, "u_bolts": {"sort": 22, "name": "U-Bolts (Bu lông chữ U)"}, "thread_rods": {"sort": 113, "name": "Thread Rods (Ty ren)"}, # === Đai ốc === "hex_nuts": {"sort": 25, "name": "Hex Nuts (Đai ốc lục giác)"}, "flange_nuts": {"sort": 122, "name": "Flanged Hex Nuts (Đai ốc mặt bích)"}, "structural_nuts": {"sort": 125, "name": "Structural Nuts (Đai ốc kết cấu)"}, "locking_nuts": {"sort": 27, "name": "Locking Nuts (Đai ốc chống xoay)"}, "acorn_nuts": {"sort": 26, "name": "Acorn Nuts (Đai ốc chụp)"}, "wing_nuts": {"sort": 36, "name": "Wing Nuts (Đai ốc cánh)"}, "coupling_nuts": {"sort": 31, "name": "Coupling Nuts (Đai ốc nối)"}, # === Vít === "machine_screws": {"sort": 38, "name": "Machine Screws (Vít máy)"}, "hex_socket_screws": {"sort": 39, "name": "Hex Socket Screws (Vít lục giác chìm)"}, "tapping_screws": {"sort": 40, "name": "Tapping Screws (Vít tự khoan)"}, "set_screws": {"sort": 42, "name": "Set Screws (Vít cấy / Vít không đầu)"}, "wood_screws": {"sort": 41, "name": "Wood Screws (Vít gỗ)"}, "drywall_screws": {"sort": 46, "name": "Drywall Screws (Vít thạch cao)"}, # === Long đền === "plain_washers": {"sort": 54, "name": "Plain Washers (Long đền phẳng)"}, "spring_washers": {"sort": 55, "name": "Spring Washers (Long đền vênh)"}, "disc_spring": {"sort": 146, "name": "Disc Spring (Long đền đĩa)"}, "toothed_washers": {"sort": 56, "name": "Toothed Lock Washers (Long đền răng)"}, # === Chốt, Đinh tán, Tắc kê === "split_pins": {"sort": 68, "name": "Split Pins (Chốt chẻ)"}, "spring_pins": {"sort": 71, "name": "Spring Pins (Chốt đàn hồi)"}, "blind_rivets": {"sort": 76, "name": "Blind Rivets (Đinh tán rút)"}, "wedge_anchors": {"sort": 98, "name": "Wedge Anchors (Tắc kê nở)"}, } # ===== Bộ lọc ===== STANDARD_TYPES = { "DIN": 4, "ISO": 5, "ASME": 3, "ANSI": 2, "EN": 8, "GB": 10, "JIS": 12, "BS": 6, "GOST": 11, "IFI": 1, "CNS": 7, "KS": 13, "NF": 14, "UNI": 15, "IS": 16, } THREAD_TYPES = { "metric": 1, # [M] Hệ mét "unc_unf": 2, # [UN] Hệ inch UNC/UNF "bsw_bsf": 3, # [BS] Hệ inch BSW/BSF "mj_unj": 4, # [MJ/UNJ] } def cache_key(prefix, value): h = hashlib.md5(str(value).encode()).hexdigest()[:12] return os.path.join(CACHE_DIR, f"{prefix}_{h}.json") def get_cached(path): if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) return None def set_cached(path, data): with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, separators=(",", ":")) # ===== Fetch + Parse tu globalfastener ===== def fetch_standards_list(sort_id, typ=None, thread=None): """Lay danh sach tieu chuan tu 1 nhom, co the loc theo typ va thread""" cache_val = f"{sort_id}_{typ}_{thread}" ck = cache_key("list", cache_val) cached = get_cached(ck) if cached: return cached url = f"{GF_BASE}/standards/?sort={sort_id}" if typ: url += f"&typ={typ}" if thread: url += f"&thread={thread}" print(f" [FETCH] {url}") resp = requests.get(url, headers=HEADERS, timeout=30) resp.raise_for_status() soup = BeautifulSoup(resp.text, "html.parser") standards = [] seen = set() for link in soup.find_all("a", href=re.compile(r"detail\.php\?sid=")): href = link.get("href", "") text = link.get_text(strip=True) if not text or "sid=" not in href: continue if text.lower() in ("view suppliers", "view", "suppliers"): continue full_url = urljoin(GF_BASE + "/standards/", href) # Extract sid sid_match = re.search(r'sid=([^&]+)', href) sid = sid_match.group(1) if sid_match else "" if full_url not in seen and sid: seen.add(full_url) standards.append({"name": text.strip(), "sid": sid, "url": full_url}) set_cached(ck, standards) return standards def parse_transposed_tables(soup): """Parse bang kich thuoc dang transpose""" tables = soup.find_all("table") if not tables: return [] table_info = [] for ti, table in enumerate(tables): rows = table.find_all("tr") if not rows: table_info.append((ti, "empty", 0, 0)) continue first_cells = rows[0].find_all(["td", "th"]) first_text = first_cells[0].get_text(strip=True) if first_cells else "" ncols = len(first_cells) nrows = len(rows) # SIZE table: cell dau chua keyword nhan dien + nhieu cot size_keywords = ("Size", "Nominal Diameter", "Screw Thread", "Thread", "Bolt Size", "Nominal Size") if ncols > 5 and any(kw in first_text for kw in size_keywords): table_info.append((ti, "size", nrows, ncols)) elif ncols <= 3 and nrows >= 5: table_info.append((ti, "label", nrows, ncols)) elif ncols >= 5 and nrows >= 3: # Value table: starts with a number, or M-size (M1.6, M2), or Φ-size cleaned_first = first_text.replace(",", ".").replace("(", "").replace(")", "") is_value = False try: float(cleaned_first) is_value = True except (ValueError, AttributeError): pass if not is_value and re.match(r'^[MΦ][\d.]+$', cleaned_first): is_value = True if is_value: table_info.append((ti, "value", nrows, ncols)) else: table_info.append((ti, "other", nrows, ncols)) else: table_info.append((ti, "other", nrows, ncols)) all_sizes_data = {} for i, (ti, ttype, nrows, ncols) in enumerate(table_info): if ttype != "size": continue size_table = tables[ti] sizes = [] size_rows = size_table.find_all("tr") for row_idx in [1, 0]: if row_idx >= len(size_rows): continue row = size_rows[row_idx] for cell in row.find_all(["td", "th"]): text = cell.get_text(strip=True) if not text or any(kw in text for kw in ("Size", "Nominal", "Diameter", "Screw", "Thread", "Bolt")): continue if len(text) > 20: continue cleaned = text.replace("(", "").replace(")", "") if re.match(r'^[ΦMd]?[\d.]+[/\d]*$', cleaned): if text not in sizes: sizes.append(text) if sizes: break if not sizes: continue label_idx = None value_idx = None for j in range(i + 1, len(table_info)): tj, tjtype, tjrows, tjcols = table_info[j] if tjtype == "label" and label_idx is None: label_idx = tj elif tjtype == "value" and value_idx is None: value_idx = tj break elif tjtype == "size": break if value_idx is None: continue param_names = [] if label_idx is not None: label_table = tables[label_idx] current_prefix = "" sub_qualifiers = ("min", "max", "Nominal Size", "max=nominal size", "Pitch") for row in label_table.find_all("tr"): cells = row.find_all(["td", "th"]) parts = [c.get_text(strip=True) for c in cells] if len(parts) >= 3: # VD: ['dw', 'Grade A', 'min'] -> "dw Grade A min" current_prefix = parts[0] param_names.append(" ".join(parts)) elif len(parts) == 2: if parts[0].lower() in ("grade a", "grade b") or parts[0] in sub_qualifiers: # VD: ['Grade B', 'min'] -> "current_prefix Grade B min" param_names.append(f"{current_prefix} {parts[0]} {parts[1]}") else: current_prefix = parts[0] param_names.append(f"{current_prefix} {parts[1]}") elif len(parts) == 1: text = parts[0] if text.lower() in ("min", "max") or text in sub_qualifiers: param_names.append(f"{current_prefix} {text}") else: param_names.append(text) current_prefix = text value_table = tables[value_idx] value_rows = value_table.find_all("tr") if len(param_names) != len(value_rows): param_names = [f"param_{idx}" for idx in range(len(value_rows))] for row_idx, value_row in enumerate(value_rows): cells = value_row.find_all(["td", "th"]) values = [c.get_text(strip=True) for c in cells] param = param_names[row_idx] if row_idx < len(param_names) else f"param_{row_idx}" for col_idx, val in enumerate(values): if col_idx < len(sizes): size_name = sizes[col_idx] if size_name not in all_sizes_data: all_sizes_data[size_name] = {"size": size_name} all_sizes_data[size_name][param] = val return list(all_sizes_data.values()) def fetch_standard_detail(sid): """Fetch + parse chi tiet 1 tieu chuan tu globalfastener""" ck = cache_key("detail", sid) cached = get_cached(ck) if cached: return cached url = f"{GF_BASE}/standards/detail.php?sid={sid}" print(f" [FETCH] {url}") resp = requests.get(url, headers=HEADERS, timeout=30) resp.raise_for_status() soup = BeautifulSoup(resp.text, "html.parser") description = "" for tag in soup.find_all(["h1", "h2", "h3"]): text = tag.get_text(strip=True) if text and len(text) > 5 and "Click" not in text: description = text break dimensions = parse_transposed_tables(soup) # Extract images (CAD drawing + product photo) images = {} for img in soup.find_all("img"): src = img.get("src", "") if not src or any(x in src.lower() for x in ["icon", "logo", "flag", "ads", "banner"]): continue # Fix URL: //img.globalfastener.com -> https://img.globalfastener.com if src.startswith("//"): src = "https:" + src elif src.startswith("/"): src = "https://img.globalfastener.com" + src if "pic_cadpara" in src and "drawing" not in images: images["drawing"] = src elif "/pic/" in src and "photo" not in images: images["photo"] = src result = { "description": description, "url": url, "dimensions": dimensions, "count": len(dimensions), "images": images, "fetched_at": time.strftime("%Y-%m-%d %H:%M") } set_cached(ck, result) return result # ===== HTTP Server ===== # ===== Storage helpers — SQLite-backed (see db.py) ===== # Keeps the same function signatures as before so request handlers are unchanged. from db import load_json_file, save_json_file def next_id(items): """Get next auto-increment id from a list of dicts with 'id' field""" if not items: return 1 return max(item.get("id", 0) for item in items) + 1 # ===== PDF Certificate Generator (edit template gốc) ===== PDF_TEMPLATE = os.path.join(BASE_DIR, "Mẫu lông đền TongMing.pdf") # Template pages theo nhóm - đúng page index trong PDF gốc TEMPLATE_PAGES = { "washer": 0, # Page 1: DIN 127 (Lock Washers) "nut": 1, # Page 2: DIN 934 (Hex Nuts) "screw": 2, # Page 3: DIN 912 (Hex Socket Screws) "bolt": 6, # Page 7: DIN 933 (Hex Bolts) } # COMMODITY -> Product Name mapping (từ 152 mẫu PDF gốc) # COMMODITY -> MECHANICAL PROPERTIES SAMPLE STANDARD (từ 152 mẫu PDF gốc) MECH_STANDARD_MAP = { "DIN127": "DIN127 1987 TYPE B", "DIN125": "DIN125-1990", "DIN9021": "DIN9021-1990", "DIN933": "ISO3506-1:2020", "DIN931": "ISO3506-1:2020", "DIN603": "ASTM F606M-2007", "DIN934": "ISO3506-2:2009", "DIN439": "ISO3506-2:2009", "DIN982": "DIN982-1987", "DIN985": "DIN985-1987", "DIN6923": "ISO3506-2:2009", "DIN912": "ISO3506-1:2020", "DIN7991": "ISO3506-1:2020", "DIN7984": "ISO3506-1:2020", "ISO7380": "ISO3506-1:2009", "ISO4014": "ISO3506-1:2020", "ISO4017": "ISO3506-1:2020", } PRODUCT_NAME_MAP = { "DIN127": "Lock Washers Type B-SquareEnds", "DIN125": "Washers Type A(Unchamfered)", "DIN9021": "Large size washers Type A", "DIN933": "Hexagon head BOLTS", "DIN931": "Hex Head Bolts with Shank", "DIN603": "Mushroom Head Square Neck Bolt", "DIN934": "Hex Nuts", "DIN439": "Hex Thin Nuts(Chamfered)Type B", "DIN982": "Hex Nylon Nuts", "DIN985": "Hex Nylon Nuts", "DIN6923": "Hex Nuts With Flange", "DIN912": "Hex Socket Screws(Knurled)", "DIN7991": "Hex Socket C.S. Head Screws", "DIN7984": "Low Head Socket Cap Screw", "ISO7380": "Button Socket Cap Screws", "ISO4014": "Hexagon Head Bolts", "ISO4017": "Hexagon Head Screws", } def find_template_page(doc, group, commodity=""): """Tìm trang template phù hợp: ưu tiên theo commodity, fallback theo group""" # Ưu tiên: tìm đúng trang có COMMODITY khớp if commodity: comm_clean = commodity.replace(" ", "").replace("-", "").upper() for i in range(min(40, len(doc))): page_text = doc[i].get_text().replace(" ", "").replace("-", "").upper() if comm_clean[:7] in page_text: # Match DIN933, DIN127... return i return TEMPLATE_PAGES.get(group, 0) def generate_cert_pdf(data): """Tạo PDF chứng chỉ bằng cách edit trực tiếp trên template gốc. Dùng đúng font SimSun từ Windows, căn đúng tọa độ gốc.""" import fitz from io import BytesIO SIMSUN_PATH = "C:/Windows/Fonts/simsun.ttc" doc = fitz.open(PDF_TEMPLATE) group = data.get("group", "washer") std_name = data.get("stdName", "") # e.g. "DIN 933 - 1987" # Xác định COMMODITY (lấy từ tiêu chuẩn user chọn) commodity = std_name.replace(" ", "").replace(" - ", "-") if std_name else "" # Chuẩn hóa: "DIN 933 - 1987" -> "DIN933-1987" import re as _re cm = _re.match(r'(DIN|ISO)\s*(\d+[\w.]*)\s*[-–]\s*(\d{4})', std_name) if cm: commodity = f"{cm.group(1)}{cm.group(2)}-{cm.group(3)}" # Product Name: tra mapping từ PDF gốc std_prefix = _re.sub(r'[-\s].*', '', commodity) # DIN933, DIN127... product_name = PRODUCT_NAME_MAP.get(std_prefix, data.get("prodName", "")) # Tìm đúng template page theo commodity tpl_page_idx = find_template_page(doc, group, commodity) # Copy 1 trang new_doc = fitz.open() new_doc.insert_pdf(doc, from_page=tpl_page_idx, to_page=tpl_page_idx) page = new_doc[0] # Dùng đúng font SimSun gốc từ Windows font = fitz.Font(fontfile=SIMSUN_PATH) # Scan tất cả spans gốc all_spans = [] for b in page.get_text("dict")["blocks"]: if "lines" not in b: continue for line in b["lines"]: for span in line["spans"]: text = span["text"].strip() if text: all_spans.append({ "text": text, "x": round(span["origin"][0], 1), "y": round(span["origin"][1], 1), "bbox": span["bbox"], "size": span["size"], }) # Danh sách: (bbox_to_redact, new_text, exact_origin_x, exact_origin_y, font_size) edits = [] replaced = set() def plan_edit(sp, new_text): key = f"{sp['bbox'][0]:.0f},{sp['bbox'][1]:.0f}" if key in replaced: return replaced.add(key) edits.append({ "bbox": sp["bbox"], "text": str(new_text), "x": sp["x"], "y": sp["y"], "size": sp["size"], }) # Surface mặc định Passivated surface_val = data.get("surface", "Passivated") or "Passivated" # Heat No: dùng chung cho HEAT NO và Heat Number (chemical) heatno_val = data.get("heatno", "") # ===== 1. Header: tìm label -> thay value ===== header_map = { "PO# NO": data.get("po", ""), "CUST P/N": data.get("custpn", data.get("sizeName", "")), "SIZE": data.get("sizeName", ""), "MATERIAL TYPE": data.get("material", ""), "HEAT NO": heatno_val, "SHIP QTY": data.get("qty", ""), "LOT NO": data.get("lotno", ""), "MFR.DATE": data.get("date", ""), "Surface": surface_val, "MARKING": data.get("marking", "NO MARK"), "COMMODITY": commodity, "Product Name": product_name, } header_spans = [s for s in all_spans if s["y"] < 250] for sp in header_spans: for label, new_val in header_map.items(): if not new_val: continue if label not in sp["text"]: continue # Tìm value: cùng y, x khác candidates = [s for s in header_spans if abs(s["y"] - sp["y"]) < 3 and s["text"] != sp["text"] and s["x"] != sp["x"]] # Lọc: lấy span gần nhất bên phải right = [s for s in candidates if s["x"] > sp["x"] + 10] left = [s for s in candidates if s["x"] < sp["x"] - 10 and not any(k in s["text"] for k in ("PURCHASER","ADDRESS","CERT","Report","MECSU"))] val_sp = min(right, key=lambda s: s["x"]) if right else (max(left, key=lambda s: s["x"]) if left else None) if val_sp: plan_edit(val_sp, new_val) break # ===== 2. Dimensional Inspection ===== dim_rows = data.get("dimRows", []) skip_texts = {"DIMENSIONAL INSPECTIONS","CHARACTERISTICS","MECHANICAL PROPERTIES", "CHEMICAL COMPOSITION%","SAMPLE STANDARD：","(mm)", "SPECIFICATION","ACTUAL RESULT","SAMPLE DETERMINANT"} # Tìm dòng characteristics char_lines = [] for sp in all_spans: if 240 < sp["y"] < 450 and sp["x"] < 200 and sp["text"] not in skip_texts and not sp["text"].startswith("SAMPLE"): char_lines.append(sp) for i, char_sp in enumerate(char_lines): if i >= len(dim_rows): break dr = dim_rows[i] cy = char_sp["y"] # Tìm spans cùng dòng same_line = [s for s in all_spans if abs(s["y"] - cy) < 2 and s["x"] > 200] offset_line = [s for s in all_spans if abs(s["y"] - (cy + 1.6)) < 2 and s["x"] > 340] # --- Xóa SPECIFICATION (x: 210-295) --- # Xác định: spec có khoảng (A / B) hay chỉ max/min spec_str = dr.get("specStr", "") spec_parts = [p.strip() for p in spec_str.replace("/", " ").split() if p.strip()] is_range = len(spec_parts) >= 2 and spec_parts[-1] not in ("min", "max") spec_spans = [s for s in same_line if 200 < s["x"] < 295] for sp in spec_spans: if sp["text"] == "<--->" and is_range: continue # Giữ mũi tên nếu là khoảng A-B elif sp["text"] == "<--->": plan_edit(sp, "") # Xóa mũi tên nếu là max/min else: plan_edit(sp, "") # Xóa giá trị cũ # --- Xóa và thay ACTUAL (x: 345-435) --- actual_spans = [s for s in offset_line if 340 < s["x"] < 440 and s["text"] not in ("<--->", "PASSED", "Qualified")] for sp in actual_spans: plan_edit(sp, "") # Xóa # Ghi spec mới tại tọa độ gốc if spec_str: if is_range: # Format: "min / max" -> min ở x=212, <---> ở x=242, max ở x=272 edits.append({"bbox": None, "text": spec_parts[0], "x": 212, "y": cy, "size": 8.0}) # Kiểm tra xem template gốc đã có <---> chưa, nếu chưa thì thêm has_arrow = any(s["text"] == "<--->" and abs(s["y"] - cy) < 2 and 230 < s["x"] < 260 for s in all_spans) if not has_arrow: edits.append({"bbox": None, "text": "<--->", "x": 241.6, "y": cy, "size": 8.0}) edits.append({"bbox": None, "text": spec_parts[1], "x": 272, "y": cy, "size": 8.0}) elif len(spec_parts) == 2 and spec_parts[1] in ("min", "max"): # Format: "14.38 min" - không có mũi tên edits.append({"bbox": None, "text": spec_str, "x": 230, "y": cy, "size": 8.0}) else: # Giá trị đơn edits.append({"bbox": None, "text": spec_str, "x": 230, "y": cy, "size": 8.0}) # Ghi actual mới - actual1=min đo (trái nhỏ), actual2=max đo (phải lớn) a1 = dr.get("actual1", "") a2 = dr.get("actual2", "") if a1 and a1 != "PASSED": edits.append({"bbox": None, "text": str(a1), "x": 354, "y": cy + 1.6, "size": 8.0}) if a2 and a2 != "PASSED": edits.append({"bbox": None, "text": str(a2), "x": 415, "y": cy + 1.6, "size": 8.0}) if a1 == "PASSED" or a2 == "PASSED": edits.append({"bbox": None, "text": "PASSED", "x": 382, "y": cy + 1.6, "size": 8.0}) # ===== 3. Chemical: đổi Heat Number (cùng giá trị với HEAT NO) ===== if heatno_val: for sp in all_spans: if sp["text"] == "Heat Number": # Tìm giá trị Heat Number cùng dòng hoặc dòng dưới (y + ~18) heat_vals = [s for s in all_spans if abs(s["y"] - (sp["y"] + 18)) < 3 and s["x"] < 120] for hv in heat_vals: plan_edit(hv, heatno_val) break # ===== 4. SAMPLE STANDARD: dim dùng commodity, mech dùng mapping riêng ===== mech_std = MECH_STANDARD_MAP.get(std_prefix, "ISO3506-1:2020") # Tìm tất cả SAMPLE STANDARD labels ss_labels = [sp for sp in all_spans if "SAMPLE STANDARD" in sp["text"]] # Phân biệt: cái nào thuộc DIMENSIONAL, cái nào thuộc MECHANICAL # MECHANICAL PROPERTIES label thường ở y > dim area mech_y = None for sp in all_spans: if sp["text"] == "MECHANICAL PROPERTIES": mech_y = sp["y"] break for ss_label in ss_labels: ss_vals = [s for s in all_spans if abs(s["y"] - ss_label["y"]) < 3 and s["x"] > ss_label["x"] + 20 and s["text"] not in ("SAMPLE STANDARD：",)] for sv in ss_vals: if "DIN" in sv["text"] or "ISO" in sv["text"] or "ASTM" in sv["text"]: if mech_y and abs(ss_label["y"] - mech_y) < 3: # MECHANICAL section -> dùng mech_std plan_edit(sv, mech_std) elif commodity: # DIMENSIONAL section -> dùng commodity plan_edit(sv, commodity) # ===== Apply redactions ===== for e in edits: if e["bbox"]: page.add_redact_annot(fitz.Rect(e["bbox"])) page.apply_redactions() # ===== Ghi text mới dùng SimSun - dùng đúng tọa độ origin gốc ===== tw = fitz.TextWriter(page.rect) for e in edits: if e["text"]: try: tw.append((e["x"], e["y"]), e["text"], font=font, fontsize=e["size"]) except Exception: pass tw.write_text(page) buf = BytesIO() new_doc.save(buf, garbage=4, deflate=True) return buf.getvalue() # ===== Excel Certificate Generator ===== def generate_cert_excel(data): """Tao file Excel chung chi kiem tra theo mau TongMing""" from openpyxl import Workbook from openpyxl.styles import Font, Alignment, Border, Side, PatternFill from io import BytesIO wb = Workbook() ws = wb.active ws.title = "Inspection Certificate" # Styles bold = Font(bold=True) bold_blue = Font(bold=True, color="1A237E", size=14) header_font = Font(bold=True, color="FFFFFF", size=11) header_fill = PatternFill(start_color="1A237E", end_color="1A237E", fill_type="solid") center = Alignment(horizontal="center", vertical="center") thin_border = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin") ) green_font = Font(bold=True, color="2E7D32") # Column widths ws.column_dimensions["A"].width = 28 ws.column_dimensions["B"].width = 20 ws.column_dimensions["C"].width = 15 ws.column_dimensions["D"].width = 15 ws.column_dimensions["E"].width = 15 # Title ws.merge_cells("A1:E1") ws["A1"] = "TONG MING ENTERPRISE CO., LTD" ws["A1"].font = bold_blue ws["A1"].alignment = center ws.merge_cells("A2:E2") ws["A2"] = "INSPECTION CERTIFICATE" ws["A2"].font = Font(bold=True, size=12) ws["A2"].alignment = center # Info section row = 4 info_fields = [ ("PURCHASER:", "MECSU JOINT STOCK COMPANY"), ("COMMODITY:", data.get("stdName", "")), ("SIZE:", data.get("sizeName", "")), ("MATERIAL TYPE:", data.get("material", "")), ("PO# NO:", data.get("po", "")), ("HEAT NO:", data.get("heatno", "")), ("LOT NO:", data.get("lotno", "")), ("SHIP QTY:", data.get("qty", "")), ("MFR. DATE:", data.get("date", "")), ("Surface:", data.get("surface", "")), ("MARKING:", data.get("marking", "NO MARK")), ] for label, value in info_fields: ws.cell(row=row, column=1, value=label).font = bold ws.cell(row=row, column=2, value=value) row += 1 # Dimensional Inspections header row += 1 ws.merge_cells(f"A{row}:E{row}") ws.cell(row=row, column=1, value="DIMENSIONAL INSPECTIONS (mm)").font = Font(bold=True, size=11, color="1A237E") row += 1 # Table header headers = ["CHARACTERISTICS", "SPECIFICATION", "ACTUAL #1", "ACTUAL #2", "RESULT"] for col, h in enumerate(headers, 1): cell = ws.cell(row=row, column=col, value=h) cell.font = header_font cell.fill = header_fill cell.alignment = center cell.border = thin_border row += 1 # Dim rows for dr in data.get("dimRows", []): ws.cell(row=row, column=1, value=dr["name"]).border = thin_border ws.cell(row=row, column=2, value=dr["specStr"]).border = thin_border ws.cell(row=row, column=2).alignment = center ws.cell(row=row, column=3, value=float(dr["actual1"])).border = thin_border ws.cell(row=row, column=3).alignment = center ws.cell(row=row, column=4, value=float(dr["actual2"])).border = thin_border ws.cell(row=row, column=4).alignment = center cell = ws.cell(row=row, column=5, value=dr["result"]) cell.font = green_font cell.alignment = center cell.border = thin_border row += 1 # Footer row += 1 ws.merge_cells(f"A{row}:E{row}") ws.cell(row=row, column=1, value="INSPECTION RESULT: SAMPLES TESTED CONFORM TO ALL OF SPECIFICATION AS ABOVE.").font = Font(size=9, italic=True) row += 1 ws.merge_cells(f"A{row}:E{row}") ws.cell(row=row, column=1, value="This Lot fit the demands of EN10204 3.1").font = Font(size=9, italic=True) # Save to bytes buf = BytesIO() wb.save(buf) return buf.getvalue() class FastenerHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=BASE_DIR, **kwargs) def do_GET(self): parsed = urlparse(self.path) path = parsed.path params = parse_qs(parsed.query) # Serve uploaded files (preview/download) if path.startswith("/uploads/"): filepath = os.path.join(BASE_DIR, path.lstrip("/")) # URL decode for Vietnamese filenames from urllib.parse import unquote filepath = unquote(filepath) if os.path.exists(filepath): with open(filepath, "rb") as f: data = f.read() ext = os.path.splitext(filepath)[1].lower() ct_map = {".pdf":"application/pdf", ".xlsx":"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".xls":"application/vnd.ms-excel", ".doc":"application/msword", ".docx":"application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".png":"image/png", ".jpg":"image/jpeg", ".jpeg":"image/jpeg", ".gif":"image/gif", ".webp":"image/webp"} ct = ct_map.get(ext, "application/octet-stream") inline_types = {".pdf", ".png", ".jpg", ".jpeg", ".gif", ".webp"} disposition = "inline" if ext in inline_types else "attachment" fname = os.path.basename(filepath) self.send_response(200) self.send_header("Content-Type", ct) self.send_header("Content-Disposition", f'{disposition}; filename="{fname}"') self.send_header("Content-Length", len(data)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) else: self.send_json({"error": "File not found"}, 404) return # API: danh sach nhom tieu chuan if path == "/api/groups": self.send_json(STANDARD_GROUPS) return # API: bo loc tieu chuan + he ren if path == "/api/filters": self.send_json({ "standard_types": STANDARD_TYPES, "thread_types": THREAD_TYPES }) return # API: danh sach tieu chuan trong 1 nhom (co filter) if path == "/api/standards": sort_id = params.get("sort", [None])[0] if not sort_id: self.send_json({"error": "Missing sort parameter"}, 400) return typ = params.get("typ", [None])[0] thread = params.get("thread", [None])[0] try: data = fetch_standards_list(int(sort_id), typ=typ, thread=thread) self.send_json(data) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: chi tiet 1 tieu chuan if path == "/api/detail": sid = params.get("sid", [None])[0] if not sid: self.send_json({"error": "Missing sid parameter"}, 400) return try: data = fetch_standard_detail(sid) self.send_json(data) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: proxy image từ globalfastener (tránh CORS) if path == "/api/image": img_url = params.get("url", [None])[0] if not img_url: self.send_json({"error": "Missing url parameter"}, 400) return try: resp = requests.get(img_url, headers=HEADERS, timeout=15) content_type = resp.headers.get("Content-Type", "image/png") self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", len(resp.content)) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "public, max-age=86400") self.end_headers() self.wfile.write(resp.content) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: xoa cache if path == "/api/clear-cache": count = 0 for f in os.listdir(CACHE_DIR): os.remove(os.path.join(CACHE_DIR, f)) count += 1 self.send_json({"cleared": count}) return # ===== Certificate Manager API ===== if path == "/api/cert/products": import cert_manager as cm products = cm.get_products() # Add group classification to each product enriched = [] for p in products: p2 = dict(p) p2['_group'] = cm.classify_product_group(p) enriched.append(p2) self.send_json(enriched) return if path == "/api/cert/config": import cert_manager as cm self.send_json(cm.get_config()) return if path == "/api/cert/stats": import cert_manager as cm self.send_json(cm.get_coverage_stats()) return if path.startswith("/api/cert/preview/"): import cert_manager as cm from urllib.parse import unquote part_number = unquote(path.replace("/api/cert/preview/", "")) cert_data = cm.build_certificate_data(part_number) self.send_json(cert_data) return if path == "/api/cert/search": import cert_manager as cm query = params.get("q", [""])[0] results = cm.search_products(query) self.send_json(results[:50]) return if path == "/api/cert/download": from urllib.parse import unquote filepath = unquote(params.get("path", [""])[0]) if os.path.exists(filepath): with open(filepath, "rb") as f: data = f.read() ext = os.path.splitext(filepath)[1].lower() ct_map = { ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".pdf": "application/pdf", } ct = ct_map.get(ext, "application/octet-stream") fname = os.path.basename(filepath) self.send_response(200) self.send_header("Content-Type", ct) self.send_header("Content-Disposition", f'attachment; filename="{fname}"') self.send_header("Content-Length", len(data)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) else: self.send_json({"error": "File not found"}, 404) return # API: Template fields config if path == "/api/cert/template-fields": all_fields = load_json_file("template_fields.json") or {} self.send_json(all_fields if all_fields else {}) return if path.startswith("/api/cert/template-fields/"): # GET specific supplier config: /api/cert/template-fields/TONGMING supplier_code = path.replace("/api/cert/template-fields/", "").strip("/") all_fields = load_json_file("template_fields.json") or {} config = all_fields.get(supplier_code, {}) if isinstance(all_fields, dict) else {} self.send_json(config) return # ===== Template Config API (GET) ===== # GET /api/template/parse?file= if path == "/api/template/parse": filepath = params.get("file", [None])[0] if not filepath: self.send_json({"error": "Missing file parameter"}, 400) return from urllib.parse import unquote filepath = unquote(filepath) # Resolve relative to BASE_DIR/templates if not os.path.isabs(filepath): filepath = os.path.join(BASE_DIR, "templates", filepath) if not os.path.exists(filepath): self.send_json({"error": f"File not found: {filepath}"}, 404) return try: import template_engine cells = template_engine.parse_template_cells(filepath) self.send_json({"cells": cells, "file": filepath}) except Exception as e: self.send_json({"error": str(e)}, 500) return # GET /api/template/configs - list all saved template configs if path == "/api/template/configs": configs_dir = os.path.join(BASE_DIR, "template_configs") os.makedirs(configs_dir, exist_ok=True) configs = [] for fn in os.listdir(configs_dir): if fn.endswith(".json"): try: with open(os.path.join(configs_dir, fn), "r", encoding="utf-8") as f: cfg = json.load(f) cfg["_filename"] = fn configs.append(cfg) except Exception: pass self.send_json(configs) return # GET /api/template/config/ if path.startswith("/api/template/config/"): supplier_code = path.replace("/api/template/config/", "").strip("/") if not supplier_code: self.send_json({"error": "Missing supplier_code"}, 400) return config_path = os.path.join(BASE_DIR, "template_configs", f"{supplier_code}.json") if os.path.exists(config_path): with open(config_path, "r", encoding="utf-8") as f: self.send_json(json.load(f)) else: self.send_json({"error": f"Config not found: {supplier_code}"}, 404) return # ===== Visual Template Builder APIs (GET) ===== # GET /api/visual-template/configs - list all visual template configs if path == "/api/visual-template/configs": data = load_json_file("visual_template_configs.json") self.send_json(data) return # GET /api/visual-template/config/{id} - get single config by ID m = re.match(r'^/api/visual-template/config/(\d+)$', path) if m: config_id = int(m.group(1)) data = load_json_file("visual_template_configs.json") config = next((c for c in data if c.get('id') == config_id), None) if config: self.send_json({"success": True, "config": config}) else: self.send_json({"error": "Config not found"}, 404) return # GET /api/visual-template/parse?file=xxx - re-parse an existing source file if path == "/api/visual-template/parse": qs = params file_path = qs.get("file", [""])[0] if not file_path: self.send_json({"error": "file parameter required"}, 400) return full_path = os.path.join(BASE_DIR, file_path) if not os.path.isabs(file_path) else file_path if not os.path.exists(full_path): self.send_json({"error": f"File not found: {file_path}"}, 404) return try: from visual_template import parse_file, detect_file_type file_type = detect_file_type(full_path) result = parse_file(full_path) response = { "success": True, "file_type": file_type, "preview_images": result.get("preview_images", []), } if file_type == "excel": response["sheets"] = result.get("sheets", []) elif file_type == "pdf": response["text_blocks"] = result.get("text_blocks", []) elif file_type == "word": response["elements"] = result.get("elements", []) self.send_json(response) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # GET /api/visual-template/characteristics?group=BOLT if path == "/api/visual-template/characteristics": group = params.get("group", [""])[0].strip().upper() if not group: self.send_json({"error": "group parameter required"}, 400) return try: config = load_json_file("cert_config.json") group_data = config.get("product_groups", {}).get(group, {}) dim_chars = group_data.get("dimensional_characteristics", []) # Get chemical specs (default to Carbon Steel) chem_data = config.get("chemical_specs", {}) material = chem_data.get("Carbon Steel", {}) if not material: for k, v in chem_data.items(): material = v break chem_elems = [] for elem_key in ['C', 'Si', 'Mn', 'P', 'S', 'Ni', 'Cr', 'Mo', 'Cu', 'N', 'V', 'Ti', 'Al', 'Fe', 'B']: elem_data = material.get(elem_key, {}) if not elem_data: continue spec_min = elem_data.get('min') spec_max = elem_data.get('max') if spec_min is not None and spec_max is not None: spec_str = f"{spec_min} - {spec_max}" elif spec_max is not None: spec_str = f"≤ {spec_max}" elif spec_min is not None: spec_str = f"≥ {spec_min}" else: continue chem_elems.append({"element": elem_key, "spec": spec_str}) self.send_json({ "group": group, "dimensional": dim_chars, "chemical": chem_elems, }) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # GET /api/visual-template/available-sizes?standard=DIN933 if path == "/api/visual-template/available-sizes": standard = params.get("standard", [""])[0].strip() if not standard: self.send_json({"error": "standard parameter required"}, 400) return try: specs = load_json_file("fastener_specs.json") sizes = [] # Search through all categories for matching standard for cat_key, cat_data in specs.items(): for std in cat_data.get("standards", []): std_name = std.get("standard", "") # Match: "DIN933" matches "DIN 933 - 1987" std_clean = re.sub(r'\s+', '', std_name).split('-')[0].upper() search_clean = re.sub(r'\s+', '', standard).upper() if search_clean in std_clean or std_clean.startswith(search_clean): dims = std.get("dimensions", []) for d in dims: size = d.get("size", "") if size and not size.startswith("("): # Skip optional sizes in parentheses sizes.append({ "size": size, "data": d, "category": cat_key, "standard_full": std_name, }) self.send_json({"sizes": sizes, "standard": standard}) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # GET /api/visual-template/generated?config_id=1 if path == "/api/visual-template/generated": config_id = params.get("config_id", [""])[0] gen_dir = os.path.join(BASE_DIR, "visual_templates", "generated") files = [] if os.path.isdir(gen_dir): for fname in sorted(os.listdir(gen_dir)): fpath = os.path.join(gen_dir, fname) if os.path.isfile(fpath): # Parse filename: configId_size_timestamp.xlsx parts = fname.rsplit('.', 1) name_part = parts[0] if parts else fname # Filter by config_id if provided if config_id and not name_part.startswith(f"cfg{config_id}_"): continue files.append({ "filename": fname, "path": f"visual_templates/generated/{fname}", "size_bytes": os.path.getsize(fpath), "created": time.strftime("%Y-%m-%d %H:%M", time.localtime(os.path.getmtime(fpath))), }) self.send_json({"files": files}) return # ===== CertManager Integrated App APIs (GET) ===== # GET /api/customers - list all customers if path == "/api/customers": data = load_json_file("customers.json") self.send_json(data) return # GET /api/dictionary - list all dictionary terms if path == "/api/dictionary": data = load_json_file("dictionary.json") self.send_json(data) return # GET /api/cert/history - list certificate generation history if path == "/api/cert/history": data = load_json_file("cert_history.json") self.send_json(data) return # GET /api/dashboard - aggregated stats if path == "/api/dashboard": products = load_json_file("products.json") customers = load_json_file("customers.json") history = load_json_file("cert_history.json") # Count template configs configs_dir = os.path.join(BASE_DIR, "template_configs") template_count = 0 if os.path.isdir(configs_dir): template_count = len([f for f in os.listdir(configs_dir) if f.endswith(".json")]) recent = sorted(history, key=lambda x: x.get("timestamp", ""), reverse=True)[:10] self.send_json({ "products": len(products), "customers": len(customers), "templates": template_count, "certs_generated": len(history), "recent_history": recent, }) return # Serve static files super().do_GET() def do_POST(self): parsed = urlparse(self.path) path = parsed.path # ===== Template Config API (POST) ===== # POST /api/template/upload - upload Excel template if path == "/api/template/upload": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) content_type = self.headers.get("Content-Type", "") boundary = content_type.split("boundary=")[-1].encode() parts = body.split(b"--" + boundary) filename = "" file_data = b"" for part in parts: if b"Content-Disposition" not in part: continue header_end = part.find(b"\r\n\r\n") if header_end < 0: continue header = part[:header_end].decode("utf-8", errors="replace") payload = part[header_end + 4:] if payload.endswith(b"\r\n"): payload = payload[:-2] if 'name="file"' in header: fn_match = re.search(r'filename="([^"]+)"', header) if fn_match: filename = fn_match.group(1) file_data = payload if not filename: self.send_json({"error": "No file uploaded"}, 400) return templates_dir = os.path.join(BASE_DIR, "templates") os.makedirs(templates_dir, exist_ok=True) safe_name = os.path.basename(filename) filepath = os.path.join(templates_dir, safe_name) # Avoid overwrite if os.path.exists(filepath): name_part, ext = os.path.splitext(safe_name) i = 1 while os.path.exists(filepath): safe_name = f"{name_part}_{i}{ext}" filepath = os.path.join(templates_dir, safe_name) i += 1 with open(filepath, "wb") as f: f.write(file_data) print(f" [TEMPLATE UPLOAD] {filepath} ({len(file_data)} bytes)") self.send_json({"success": True, "filename": safe_name, "path": filepath}) return # POST /api/template/config - save a template config if path == "/api/template/config": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) supplier_code = data.get("supplier_code", "") if not supplier_code: self.send_json({"error": "supplier_code is required"}, 400) return configs_dir = os.path.join(BASE_DIR, "template_configs") os.makedirs(configs_dir, exist_ok=True) config_path = os.path.join(configs_dir, f"{supplier_code}.json") with open(config_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f" [TEMPLATE CONFIG] Saved {config_path}") self.send_json({"success": True, "supplier_code": supplier_code, "path": config_path}) return # POST /api/template/generate - generate certificate from template if path == "/api/template/generate": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) part_number = data.get("part_number", "") supplier_code = data.get("supplier_code", "") overrides = data.get("overrides", {}) if not part_number or not supplier_code: self.send_json({"error": "part_number and supplier_code are required"}, 400) return try: # Load template config config_path = os.path.join(BASE_DIR, "template_configs", f"{supplier_code}.json") if not os.path.exists(config_path): self.send_json({"error": f"Template config not found: {supplier_code}"}, 404) return with open(config_path, "r", encoding="utf-8") as f: config = json.load(f) # Load template file template_file = config.get("template_file", "") if not template_file: self.send_json({"error": "No template_file in config"}, 400) return if not os.path.isabs(template_file): template_file = os.path.join(BASE_DIR, "templates", template_file) if not os.path.exists(template_file): self.send_json({"error": f"Template file not found: {template_file}"}, 404) return # Get cert data from cert_manager import cert_manager as cm cert_data = cm.build_certificate_data(part_number, overrides=overrides) if cert_data.get("error"): self.send_json(cert_data, 400) return cert_data["_overrides"] = overrides # Generate output path certs_dir = os.path.join(BASE_DIR, "certificates") os.makedirs(certs_dir, exist_ok=True) timestamp = time.strftime("%Y%m%d_%H%M%S") safe_part = re.sub(r'[^\w\-.]', '_', part_number) output_filename = f"{supplier_code}_{safe_part}_{timestamp}.xlsx" output_path = os.path.join(certs_dir, output_filename) # Fill template import template_engine result = template_engine.fill_template(template_file, config, cert_data, output_path) self.send_json(result) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # API: Save template fields config if path == "/api/cert/template-fields": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) supplier_code = data.get("supplier_code", "") config = data.get("config", {}) if not supplier_code: self.send_json({"error": "supplier_code required"}, 400) return all_fields = load_json_file("template_fields.json") or {} if not isinstance(all_fields, dict): all_fields = {} all_fields[supplier_code] = config save_json_file("template_fields.json", all_fields) self.send_json({"success": True, "supplier_code": supplier_code}) return # API: Certificate Manager - Generate certificate if path == "/api/cert/generate": import cert_manager as cm content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) try: result = cm.generate_certificate( data.get("part_number", ""), output_format=data.get("format", "mecsu_iqc"), overrides=data.get("overrides", {}) ) # Log to cert_history if result.get("success") or result.get("path"): history = load_json_file("cert_history.json") history_entry = { "id": next_id(history), "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "date": time.strftime("%Y-%m-%d"), "part_number": data.get("part_number", ""), "description": data.get("description", ""), "format": data.get("format", "mecsu_iqc"), "overrides": data.get("overrides", {}), "output_file": result.get("path", ""), } history.append(history_entry) save_json_file("cert_history.json", history) self.send_json(result) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: Generate + Preview (generate cert, optionally convert DOCX→PDF, return file path) if path == "/api/cert/preview-file": import cert_manager as cm content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) try: fmt = data.get("format", "tongming") result = cm.generate_certificate( data.get("part_number", ""), output_format=fmt, overrides=data.get("overrides", {}) ) if result.get("success") and result.get("path"): filepath = result["path"] pdf_path = filepath # Convert DOCX to PDF for preview if filepath.endswith('.docx'): try: from docx2pdf import convert pdf_path = filepath.replace('.docx', '.pdf') convert(filepath, pdf_path) except Exception as conv_err: # Fallback: serve DOCX as-is print(f" [WARN] DOCX→PDF convert failed: {conv_err}") pdf_path = filepath # Log to cert_history try: history = load_json_file("cert_history.json") history_entry = { "id": next_id(history), "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "date": time.strftime("%Y-%m-%d"), "part_number": data.get("part_number", ""), "description": data.get("description", ""), "format": fmt, "overrides": data.get("overrides", {}), "output_file": pdf_path, } history.append(history_entry) save_json_file("cert_history.json", history) except Exception as hist_err: print(f" [WARN] History log failed: {hist_err}") self.send_json({"success": True, "path": pdf_path, "docx_path": filepath, "format": fmt}) else: self.send_json(result) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: export certificate to PDF (edit template gốc) if path == "/api/export-cert-pdf": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) try: pdf_bytes = generate_cert_pdf(data) self.send_response(200) self.send_header("Content-Type", "application/pdf") self.send_header("Content-Disposition", "attachment; filename=certificate.pdf") self.send_header("Content-Length", len(pdf_bytes)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(pdf_bytes) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # API: export certificate to Excel if path == "/api/export-cert": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) try: excel_bytes = generate_cert_excel(data) self.send_response(200) self.send_header("Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") self.send_header("Content-Disposition", "attachment; filename=certificate.xlsx") self.send_header("Content-Length", len(excel_bytes)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(excel_bytes) except Exception as e: self.send_json({"error": str(e)}, 500) return # API: upload template/customer file if path == "/api/upload-file": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) content_type = self.headers.get("Content-Type", "") # Parse multipart form data manually boundary = content_type.split("boundary=")[-1].encode() parts = body.split(b"--" + boundary) filename = "" file_data = b"" folder = "templates" for part in parts: if b"Content-Disposition" not in part: continue header_end = part.find(b"\r\n\r\n") if header_end < 0: continue header = part[:header_end].decode("utf-8", errors="replace") payload = part[header_end + 4:] # Remove trailing \r\n-- if payload.endswith(b"\r\n"): payload = payload[:-2] if 'name="folder"' in header: folder = payload.decode("utf-8").strip() elif 'name="file"' in header: # Extract filename fn_match = re.search(r'filename="([^"]+)"', header) if fn_match: filename = fn_match.group(1) file_data = payload if not filename: self.send_json({"error": "No file uploaded"}, 400) return upload_dir = os.path.join(BASE_DIR, "uploads", folder) os.makedirs(upload_dir, exist_ok=True) safe_name = os.path.basename(filename) filepath = os.path.join(upload_dir, safe_name) if os.path.exists(filepath): name_part, ext = os.path.splitext(safe_name) i = 1 while os.path.exists(filepath): safe_name = f"{name_part}_{i}{ext}" filepath = os.path.join(upload_dir, safe_name) i += 1 with open(filepath, "wb") as f: f.write(file_data) print(f" [UPLOAD] {filepath} ({len(file_data)} bytes)") self.send_json({"success": True, "filename": safe_name, "path": f"uploads/{folder}/{safe_name}"}) return # API: serve uploaded file (inline preview or download) if path.startswith("/uploads/"): filepath = os.path.join(BASE_DIR, path.lstrip("/")) if os.path.exists(filepath): with open(filepath, "rb") as f: data = f.read() ext = os.path.splitext(filepath)[1].lower() ct_map = {".pdf":"application/pdf", ".xlsx":"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ".xls":"application/vnd.ms-excel", ".doc":"application/msword", ".docx":"application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".png":"image/png", ".jpg":"image/jpeg", ".jpeg":"image/jpeg", ".gif":"image/gif", ".webp":"image/webp"} ct = ct_map.get(ext, "application/octet-stream") # Inline for viewable types, attachment for others inline_types = {".pdf", ".png", ".jpg", ".jpeg", ".gif", ".webp"} disposition = "inline" if ext in inline_types else "attachment" fname = os.path.basename(filepath) self.send_response(200) self.send_header("Content-Type", ct) self.send_header("Content-Disposition", f"{disposition}; filename=\"{fname}\"") self.send_header("Content-Length", len(data)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) else: self.send_json({"error": "File not found"}, 404) return # ===== Visual Template Builder APIs (POST) ===== # POST /api/visual-template/upload - upload file and convert to images if path == "/api/visual-template/upload": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) content_type = self.headers.get("Content-Type", "") boundary = content_type.split("boundary=")[-1].encode() parts = body.split(b"--" + boundary) filename = "" file_data = b"" for part in parts: if b"Content-Disposition" not in part: continue header_end = part.find(b"\r\n\r\n") if header_end < 0: continue header = part[:header_end].decode("utf-8", errors="replace") payload = part[header_end + 4:] if payload.endswith(b"\r\n"): payload = payload[:-2] if 'name="file"' in header: fn_match = re.search(r'filename="([^"]+)"', header) if fn_match: filename = fn_match.group(1) file_data = payload if not filename: self.send_json({"error": "No file uploaded"}, 400) return vt_dir = os.path.join(BASE_DIR, "visual_templates") os.makedirs(vt_dir, exist_ok=True) safe_name = os.path.basename(filename) filepath = os.path.join(vt_dir, safe_name) # Avoid overwrite if os.path.exists(filepath): name_part, ext = os.path.splitext(safe_name) i = 1 while os.path.exists(filepath): safe_name = f"{name_part}_{i}{ext}" filepath = os.path.join(vt_dir, safe_name) i += 1 with open(filepath, "wb") as f: f.write(file_data) print(f" [VT UPLOAD] {filepath} ({len(file_data)} bytes)") try: from visual_template import parse_file, detect_file_type # Validate file type file_type = detect_file_type(filepath) # Parse file structure result = parse_file(filepath) # Copy source file to templates/files/ for later use files_dir = os.path.join(BASE_DIR, "visual_templates", "files") os.makedirs(files_dir, exist_ok=True) import shutil stored_path = os.path.join(files_dir, safe_name) if filepath != stored_path: shutil.copy2(filepath, stored_path) # Build response response = { "success": True, "source_file": os.path.join("visual_templates", "files", safe_name).replace("\\", "/"), "file_type": file_type, "preview_images": result.get("preview_images", []), } # Include structured data based on file type if file_type == "excel": response["sheets"] = result.get("sheets", []) elif file_type == "pdf": response["text_blocks"] = result.get("text_blocks", []) elif file_type == "word": response["elements"] = result.get("elements", []) self.send_json(response) except ValueError as ve: self.send_json({"error": str(ve)}, 400) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # POST /api/visual-template/check-duplicate - check if template exists for NCC+Standard+Group if path == "/api/visual-template/check-duplicate": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) supplier = data.get("supplier", "").strip().lower() standard = data.get("standard", "").strip().lower() product_group = data.get("product_group", "").strip().lower() configs = load_json_file("visual_template_configs.json") existing = None for c in configs: if (c.get("supplier", "").strip().lower() == supplier and c.get("standard", "").strip().lower() == standard and c.get("product_group", "").strip().lower() == product_group): existing = c break if existing: self.send_json({"exists": True, "existing": existing}) else: self.send_json({"exists": False}) return # POST /api/visual-template/config - save a visual template config if path == "/api/visual-template/config": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) configs = load_json_file("visual_template_configs.json") data["id"] = next_id(configs) data["created"] = time.strftime("%Y-%m-%d") configs.append(data) save_json_file("visual_template_configs.json", configs) print(f" [VT CONFIG] Saved config id={data['id']} name={data.get('name','')}") self.send_json({"success": True, "config": data}) return # POST /api/visual-template/generate - generate file from template if path == "/api/visual-template/generate": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) template_id = data.get("template_id") field_values = data.get("field_values", {}) if not template_id: self.send_json({"error": "template_id is required"}, 400) return try: from visual_template import generate_from_template output_path = generate_from_template(template_id, field_values) from urllib.parse import quote # Determine output file type for download ext = os.path.splitext(output_path)[1].lower() file_type = "excel" if ext in (".xlsx", ".xls") else "pdf" if ext == ".pdf" else "word" # Log to cert_history try: history = load_json_file("cert_history.json") history_entry = { "id": next_id(history), "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "date": time.strftime("%Y-%m-%d"), "part_number": field_values.get("product_name", "") or field_values.get("part_number", ""), "description": field_values.get("product_name", ""), "format": "vt_" + str(template_id), "output_file": output_path, "file_type": file_type, } history.append(history_entry) save_json_file("cert_history.json", history) except Exception as hist_err: print(f" [WARN] VT History log failed: {hist_err}") self.send_json({ "success": True, "download_url": "/api/cert/download?path=" + quote(output_path, safe=''), "path": output_path, "file_type": file_type, }) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # POST /api/visual-template/generate-batch - generate templates for multiple sizes if path == "/api/visual-template/generate-batch": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) config_id = data.get("config_id") sizes = data.get("sizes", []) # list of {size, data} objects if not config_id or not sizes: self.send_json({"error": "config_id and sizes required"}, 400) return try: from visual_template import generate_batch_from_template results = generate_batch_from_template(config_id, sizes) self.send_json({"success": True, "results": results}) except Exception as e: import traceback traceback.print_exc() self.send_json({"error": str(e)}, 500) return # ===== CertManager Integrated App APIs (POST) ===== # POST /api/customers - add new customer if path == "/api/customers": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) customers = load_json_file("customers.json") data["id"] = next_id(customers) customers.append(data) save_json_file("customers.json", customers) self.send_json(customers) return # POST /api/dictionary - add new dictionary term if path == "/api/dictionary": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) dictionary = load_json_file("dictionary.json") data["id"] = next_id(dictionary) dictionary.append(data) save_json_file("dictionary.json", dictionary) self.send_json(dictionary) return # POST /api/cert/products/add - add single product if path == "/api/cert/products/add": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) products = load_json_file("products.json") # Check duplicate part number part = data.get("part", "") if any(p.get("part") == part for p in products): self.send_json({"error": f"Product with part '{part}' already exists"}, 400) return products.append(data) save_json_file("products.json", products) self.send_json({"success": True, "count": len(products)}) return # API: upload products Excel if path == "/api/upload-products": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) content_type = self.headers.get("Content-Type", "") boundary = content_type.split("boundary=")[-1].encode() parts = body.split(b"--" + boundary) file_data = b"" for part in parts: if b'name="file"' in part and b"Content-Disposition" in part: header_end = part.find(b"\r\n\r\n") if header_end >= 0: file_data = part[header_end + 4:] if file_data.endswith(b"\r\n"): file_data = file_data[:-2] if file_data: import pandas as pd from io import BytesIO try: df = pd.read_excel(BytesIO(file_data)) products = self._parse_products_df(df) save_json_file("products.json", products) self.send_json({"count": len(products), "status": "ok"}) except Exception as e: self.send_json({"error": str(e)}, 500) else: self.send_json({"error": "No file"}, 400) return @staticmethod def _parse_products_df(df): """Parse products DataFrame, extract standard and size""" import re # Try to find Part# and Description columns part_col = None desc_col = None cate_col = None for c in df.columns: cl = str(c).lower().strip() if cl in ("part#", "part", "part_number", "partnumber", "mã hàng"): part_col = c elif cl in ("description", "part_description", "tên mã hàng", "mô tả"): desc_col = c elif cl in ("category_name", "cate", "phân loại"): cate_col = c if not part_col or not desc_col: # Fallback: first 2 columns cols = list(df.columns) part_col = cols[0] desc_col = cols[1] if len(cols) > 1 else cols[0] products = [] seen = set() for _, row in df.iterrows(): part = str(row.get(part_col, "")).strip() if not part or part in seen or part == "nan": continue seen.add(part) desc = str(row.get(desc_col, "")).strip() cat = str(row.get(cate_col, "")).strip() if cate_col else "" if cat == "nan": cat = "" # Extract standard m = re.search(r"(DIN\s*\d+[\w]*|ISO\s*\d+[\w.-]*)", desc, re.IGNORECASE) standard = m.group(1).strip() if m else "" # Extract size m = re.search(r"\b(M\d+(?:[xX]\d+)?)\b", desc) if m: size = m.group(1) else: m = re.search(r"OD([\d.]+[xX][\d.]+)\s*mm", desc) size = "OD" + m.group(1) + "mm" if m else "" products.append({ "part": part, "description": desc, "standard": standard, "size": size, "category_name": cat, "category_name__1": "" }) return products def do_PUT(self): parsed = urlparse(self.path) path = parsed.path content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) data = json.loads(body.decode("utf-8")) # PUT /api/customers/{id} - update customer m = re.match(r'^/api/customers/(\d+)$', path) if m: cust_id = int(m.group(1)) customers = load_json_file("customers.json") found = False for i, c in enumerate(customers): if c.get("id") == cust_id: data["id"] = cust_id customers[i] = data found = True break if not found: self.send_json({"error": f"Customer {cust_id} not found"}, 404) return save_json_file("customers.json", customers) self.send_json(customers) return # PUT /api/dictionary/{id} - update dictionary term m = re.match(r'^/api/dictionary/(\d+)$', path) if m: term_id = int(m.group(1)) dictionary = load_json_file("dictionary.json") found = False for i, t in enumerate(dictionary): if t.get("id") == term_id: data["id"] = term_id dictionary[i] = data found = True break if not found: self.send_json({"error": f"Term {term_id} not found"}, 404) return save_json_file("dictionary.json", dictionary) self.send_json(dictionary) return # PUT /api/visual-template/config/{id} - update visual template config m = re.match(r'^/api/visual-template/config/(\d+)$', path) if m: config_id = int(m.group(1)) configs = load_json_file("visual_template_configs.json") found = False for i, c in enumerate(configs): if c.get("id") == config_id: data["id"] = config_id data["updated"] = time.strftime("%Y-%m-%d") configs[i] = data found = True break if not found: self.send_json({"error": f"Visual template config {config_id} not found"}, 404) return save_json_file("visual_template_configs.json", configs) self.send_json({"success": True, "config": data}) return # PUT /api/cert/products/{part} - update single product if path.startswith("/api/cert/products/"): from urllib.parse import unquote part_number = unquote(path.replace("/api/cert/products/", "")) products = load_json_file("products.json") found = False for i, p in enumerate(products): if p.get("part") == part_number: products[i] = data found = True break if not found: self.send_json({"error": f"Product '{part_number}' not found"}, 404) return save_json_file("products.json", products) self.send_json({"success": True, "count": len(products)}) return self.send_json({"error": "Not found"}, 404) def do_DELETE(self): parsed = urlparse(self.path) path = parsed.path # DELETE /api/customers/{id} - delete customer m = re.match(r'^/api/customers/(\d+)$', path) if m: cust_id = int(m.group(1)) customers = load_json_file("customers.json") new_customers = [c for c in customers if c.get("id") != cust_id] if len(new_customers) == len(customers): self.send_json({"error": f"Customer {cust_id} not found"}, 404) return save_json_file("customers.json", new_customers) self.send_json(new_customers) return # DELETE /api/dictionary/{id} - delete dictionary term m = re.match(r'^/api/dictionary/(\d+)$', path) if m: term_id = int(m.group(1)) dictionary = load_json_file("dictionary.json") new_dict = [t for t in dictionary if t.get("id") != term_id] if len(new_dict) == len(dictionary): self.send_json({"error": f"Term {term_id} not found"}, 404) return save_json_file("dictionary.json", new_dict) self.send_json(new_dict) return # DELETE /api/visual-template/config/{id} - delete visual template config m = re.match(r'^/api/visual-template/config/(\d+)$', path) if m: config_id = int(m.group(1)) configs = load_json_file("visual_template_configs.json") deleted_config = None new_configs = [] for c in configs: if c.get("id") == config_id: deleted_config = c else: new_configs.append(c) if not deleted_config: self.send_json({"error": f"Visual template config {config_id} not found"}, 404) return # Delete source file if exists source_file = deleted_config.get("source_file", "") if source_file: full_path = os.path.join(BASE_DIR, source_file) if not os.path.isabs(source_file) else source_file if os.path.exists(full_path): try: os.remove(full_path) print(f" [VT DELETE] Removed file: {full_path}") except Exception as e: print(f" [VT DELETE] Could not remove file: {e}") save_json_file("visual_template_configs.json", new_configs) print(f" [VT DELETE] Removed config id={config_id} name={deleted_config.get('name','')}") self.send_json({"success": True, "configs": new_configs}) return # DELETE /api/visual-template/generated/{filename} - delete a generated file m = re.match(r'^/api/visual-template/generated/(.+)$', path) if m: from urllib.parse import unquote filename = unquote(m.group(1)) gen_dir = os.path.join(BASE_DIR, "visual_templates", "generated") filepath = os.path.join(gen_dir, filename) if os.path.exists(filepath): os.remove(filepath) print(f" [VT DELETE] Removed generated file: {filename}") self.send_json({"success": True}) else: self.send_json({"error": "File not found"}, 404) return # DELETE /api/cert/products/{part} - delete single product if path.startswith("/api/cert/products/"): from urllib.parse import unquote part_number = unquote(path.replace("/api/cert/products/", "")) products = load_json_file("products.json") new_products = [p for p in products if p.get("part") != part_number] if len(new_products) == len(products): self.send_json({"error": f"Product '{part_number}' not found"}, 404) return save_json_file("products.json", new_products) self.send_json({"success": True, "count": len(new_products)}) return self.send_json({"error": "Not found"}, 404) def do_OPTIONS(self): """Handle CORS preflight""" self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def _excel_to_html(self, filepath): """Convert Excel file to HTML table for browser preview.""" from openpyxl import load_workbook from openpyxl.utils import get_column_letter wb = load_workbook(filepath) ws = wb.active html = '
' html += f'
Preview: {os.path.basename(filepath)}
' html += '' for row in ws.iter_rows(min_row=1, max_row=ws.max_row): html += '' for cell in row: val = str(cell.value) if cell.value is not None else '' # Style from cell bold = 'font-weight:bold;' if (cell.font and cell.font.bold) else '' bg = '' if cell.fill and cell.fill.start_color and cell.fill.start_color.rgb and cell.fill.start_color.rgb != '00000000': try: rgb = str(cell.fill.start_color.rgb) if len(rgb) == 8: rgb = rgb[2:] # Remove alpha if rgb and rgb != '000000': bg = f'background:#{rgb};' except: pass align = 'text-align:center;' if (cell.alignment and cell.alignment.horizontal == 'center') else '' border = 'border:1px solid #ddd;' fg = '' if cell.font and cell.font.color and cell.font.color.rgb: try: frgb = str(cell.font.color.rgb) if len(frgb) == 8: frgb = frgb[2:] if frgb and frgb not in ('000000', '00000000'): fg = f'color:#{frgb};' except: pass style = f'{bold}{bg}{align}{border}{fg}padding:3px 5px;' # Handle merged cells html += f'' html += '' html += '
{val}
' return html def _docx_to_html(self, filepath): """Convert DOCX to simple HTML preview.""" from docx import Document doc = Document(filepath) html = '
' html += f'
Preview: {os.path.basename(filepath)}
' for p in doc.paragraphs: if p.text.strip(): bold = 'font-weight:bold;' if any(r.bold for r in p.runs) else '' size = '' for r in p.runs: if r.font.size: pt = r.font.size.pt size = f'font-size:{pt}px;' break html += f'

{p.text}

' for table in doc.tables: html += '' for row in table.rows: html += '' seen = set() for cell in row.cells: cell_id = id(cell._tc) if cell_id in seen: continue seen.add(cell_id) val = cell.text.strip() html += f'' html += '' html += '
{val}
' html += '
' return html def send_json(self, data, status=200): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", len(body)) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def log_message(self, format, *args): if "/api/" in str(args[0]): print(f" [API] {args[0]}") if __name__ == "__main__": print("=" * 50) print(f" Fastener Spec Server + Certificate Manager") print(f" http://localhost:{PORT}/prototype.html") print(f" http://localhost:{PORT}/cert_manager.html") print("=" * 50) print(f" Cache: {CACHE_DIR}") print() print("API endpoints:") print(f" GET /api/groups -> danh sach nhom") print(f" GET /api/standards?sort=55 -> tieu chuan trong nhom") print(f" GET /api/detail?sid=xxx -> thong so chi tiet") print(f" GET /api/clear-cache -> xoa cache") print(f" GET /api/cert/products -> danh sach san pham (cert)") print(f" GET /api/cert/config -> bo tieu chuan") print(f" GET /api/cert/stats -> thong ke coverage") print(f" GET /api/cert/preview/{{part}} -> xem truoc chung chi") print(f" POST /api/cert/generate -> tao chung chi (+ log history)") print(f" GET /api/template/parse -> parse Excel template cells") print(f" GET /api/template/configs -> list all template configs") print(f" GET /api/template/config/X -> get specific config") print(f" POST /api/template/upload -> upload Excel template") print(f" POST /api/template/config -> save template config") print(f" POST /api/template/generate -> generate cert from template") print(f" --- Visual Template Builder ---") print(f" POST /api/visual-template/upload -> upload & convert to images") print(f" GET /api/visual-template/configs -> list visual template configs") print(f" POST /api/visual-template/config -> save visual template config") print(f" PUT /api/visual-template/config/{{id}} -> update visual template config") print(f" DELETE /api/visual-template/config/{{id}} -> delete visual template config") print(f" POST /api/visual-template/generate -> generate filled PDF") print(f" --- CertManager Integrated ---") print(f" GET /api/customers -> danh sach khach hang") print(f" POST /api/customers -> them khach hang") print(f" PUT /api/customers/{{id}} -> cap nhat khach hang") print(f" DELETE /api/customers/{{id}} -> xoa khach hang") print(f" GET /api/dictionary -> tu dien thuat ngu") print(f" POST /api/dictionary -> them thuat ngu") print(f" PUT /api/dictionary/{{id}} -> cap nhat thuat ngu") print(f" DELETE /api/dictionary/{{id}} -> xoa thuat ngu") print(f" GET /api/cert/history -> lich su tao chung chi") print(f" GET /api/dashboard -> thong ke tong hop") print(f" POST /api/cert/products/add -> them san pham") print(f" PUT /api/cert/products/{{part}} -> cap nhat san pham") print(f" DELETE /api/cert/products/{{part}} -> xoa san pham") print() # Initialize SQLite storage (auto-migrate JSON files on first run) import db db._ensure_init() print(f" [db] SQLite ready: {db.DB_PATH}") print(f" [db] keys: {', '.join(db.list_keys())}") print() server = http.server.HTTPServer(("0.0.0.0", PORT), FastenerHandler) # Lấy IP local để hiện link network import socket try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) local_ip = s.getsockname()[0] s.close() except Exception: local_ip = "localhost" print(f"Server running on:") print(f" Local: http://localhost:{PORT}/prototype.html") print(f" Network: http://{local_ip}:{PORT}/prototype.html") print("Nhan Ctrl+C de dung.\n") try: server.serve_forever() except KeyboardInterrupt: print("\nServer stopped.") server.server_close()