Create app.py

app.py

@@ -0,0 +1,282 @@
+import os
+import io
+import csv
+import time
+import threading
+from datetime import datetime, timezone
+from typing import Dict, Optional, Tuple
+
+import gradio as gr
+from huggingface_hub import HfApi
+
+# -------------------------
+# Config
+# -------------------------
+DATASET_REPO_ID = os.environ.get("DATASET_REPO_ID") # Space secret
+HF_TOKEN = os.environ["HF_TOKEN"]          # Space secret
+
+ORDERS_PATH = "orders.csv"
+SENT_PATH = "sent.csv"
+KEYS_PATH = "keys.csv"
+
+# # Optional SMTP config (Space secrets)
+# SMTP_HOST = os.getenv("SMTP_HOST")
+# SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
+# SMTP_USER = os.getenv("SMTP_USER")
+# SMTP_PASS = os.getenv("SMTP_PASS")
+# FROM_EMAIL = os.getenv("FROM_EMAIL", SMTP_USER or "no-reply@example.com")
+
+api = HfApi(token=HF_TOKEN)
+_lock = threading.Lock()
+
+# Simple cache to reduce Hub reads
+_cache = {
+    "ts": 0.0,
+    "orders": None,  # type: ignore
+    "sent": None,  # type: ignore
+    "keys": None,  # type: ignore
+}
+CACHE_TTL_SEC = 10.0
+
+# -------------------------
+# Helpers
+# -------------------------
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
+
+def hash_order(order_number: str) -> str:
+    normalized = order_number.strip().replace(" ", "")
+    return normalized
+
+def _download_csv(repo_id: str, path_in_repo: str) -> str:
+    # Download file bytes from the dataset repo, return text
+    file_bytes = api.hf_hub_download(
+        repo_id=repo_id,
+        filename=path_in_repo,
+        repo_type="dataset",
+        token=HF_TOKEN,
+    )
+    with open(file_bytes, "r", encoding="utf-8") as f:
+        return f.read()
+
+def _upload_csv(repo_id: str, path_in_repo: str, content: str, commit_message: str) -> None:
+    api.upload_file(
+        path_or_fileobj=io.BytesIO(content.encode("utf-8")),
+        path_in_repo=path_in_repo,
+        repo_id=repo_id,
+        repo_type="dataset",
+        commit_message=commit_message,
+        token=HF_TOKEN,
+    )
+
+def _parse_orders(csv_text: str) -> Dict[str, Dict[str, str]]:
+    rows: Dict[str, Dict[str, str]] = {}
+    reader = csv.DictReader(io.StringIO(csv_text))
+    for r in reader:
+        rows[r["order_number"]] = {
+            "order_number": r["order_number"],
+            "contact_person": r["contact_person"],
+            "email_address": r["email_address"],
+            "quantity": r["quantity"],
+        }
+    return rows
+
+def _parse_sent(csv_text: str) -> Dict[str, Dict[str, str]]:
+    rows: Dict[str, Dict[str, str]] = {}
+    reader = csv.DictReader(io.StringIO(csv_text))
+    for r in reader:
+        rows[r["order_number"]] = {
+            "order_number": r["order_number"],
+            "quantity": r["quantity"],
+            "keys_sent": (r.get("keys_sent") or "").strip(),
+            "quantity_keys_sent": (r.get("quantity_keys_sent") or "").strip(),
+        }
+    return rows
+
+def _parse_keys(csv_text: str) -> list:
+    keys = []
+    reader = csv.DictReader(io.StringIO(csv_text))
+    for r in reader:
+        if r["key"].strip():
+            keys.append({
+                "key": r["key"].strip(),
+                "sent": (r.get("sent") or "").strip(),
+            })
+    return keys
+
+def _serialize_sent(rows: Dict[str, Dict[str, str]]) -> str:
+    out = io.StringIO()
+    fieldnames = ["order_number", "quantity", "keys_sent", "quantity_keys_sent"]
+    w = csv.DictWriter(out, fieldnames=fieldnames)
+    w.writeheader()
+    for _, r in rows.items():
+        w.writerow({
+            "order_number": r["order_number"],
+            "quantity": r["quantity"],
+            "keys_sent": r.get("keys_sent", ""),
+            "quantity_keys_sent": r.get("quantity_keys_sent", ""),
+        })
+    return out.getvalue()
+
+def _serialize_keys(keys: list) -> str:
+    out = io.StringIO()
+    fieldnames = ["key", "sent"]
+    w = csv.DictWriter(out, fieldnames=fieldnames)
+    w.writeheader()
+    for k in keys:
+        w.writerow({
+            "key": k["key"],
+            "sent": k.get("sent", ""),
+        })
+    return out.getvalue()
+
+def load_state(force: bool = False):
+    now = time.time()
+    if (not force) and _cache["orders"] is not None and (now - _cache["ts"] < CACHE_TTL_SEC):
+        return _cache["orders"], _cache["sent"], _cache["keys"]
+
+    orders_csv = _download_csv(DATASET_REPO_ID, ORDERS_PATH)
+    sent_csv = _download_csv(DATASET_REPO_ID, SENT_PATH)
+    keys_csv = _download_csv(DATASET_REPO_ID, KEYS_PATH)
+
+    orders = _parse_orders(orders_csv)
+    sent = _parse_sent(sent_csv)
+    keys = _parse_keys(keys_csv)
+
+    _cache["ts"] = now
+    _cache["orders"] = orders
+    _cache["sent"] = sent
+    _cache["keys"] = keys
+    return orders, sent, keys
+
+def send_key_email(to_email: str, keys: list, contact_person: str) -> None:
+    """
+    Send keys to the customer via email.
+    """
+    # if not (SMTP_HOST and SMTP_USER and SMTP_PASS):
+    #     return
+
+    # import smtplib
+    # from email.message import EmailMessage
+
+    # msg = EmailMessage()
+    # msg["Subject"] = "Your Gradium Keys"
+    # msg["From"] = FROM_EMAIL
+    # msg["To"] = to_email
+    # keys_text = "\n".join([f"- {k}" for k in keys])
+    # msg.set_content(
+    #     f"Hi {contact_person},\n\n"
+    #     f"Here are your Gradium keys:\n\n"
+    #     f"{keys_text}\n\n"
+    #     f"Best regards,\nGradium Team\n"
+    # )
+
+    # with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as s:
+    #     s.starttls()
+    #     s.login(SMTP_USER, SMTP_PASS)
+    #     s.send_message(msg)
+
+    pass
+
+# -------------------------
+# Core API
+# -------------------------
+def send_keys_for_order(
+    order_number: str,
+    request: Optional[gr.Request] = None,
+) -> Tuple[str, str]:
+    """
+    Returns (keys_sent, status_message)
+    """
+    if not order_number or not order_number.strip():
+        return "", "Missing order number."
+
+    order_number = order_number.strip()
+
+    _ = request
+
+    with _lock:
+        orders, sent, keys = load_state(force=True)
+
+        order = orders.get(order_number)
+        if order is None:
+            return "", f"Order number '{order_number}' not found."
+
+        sent_info = sent.get(order_number)
+        if sent_info is None:
+            return "", f"Order '{order_number}' not found in sent.csv."
+
+        quantity_needed = int(order["quantity"])
+
+        # Check if already sent
+        if sent_info["keys_sent"]:
+            return sent_info["keys_sent"], f"Keys already sent: {sent_info['keys_sent']}"
+
+        # Find available keys
+        available_keys = [k for k in keys if not k["sent"]]
+
+        if len(available_keys) < quantity_needed:
+            return "", f"Not enough available keys. Need {quantity_needed}, have {len(available_keys)}."
+
+        # Allocate keys
+        keys_to_send = available_keys[:quantity_needed]
+        keys_str = ",".join([k["key"] for k in keys_to_send])
+
+        # Mark keys as sent
+        for k in keys_to_send:
+            k["sent"] = _utc_now_iso()
+
+        # Update sent.csv
+        sent_info["keys_sent"] = keys_str
+        sent_info["quantity_keys_sent"] = str(quantity_needed)
+
+        # Save changes
+        updated_sent_csv = _serialize_sent(sent)
+        updated_keys_csv = _serialize_keys(keys)
+
+        _upload_csv(
+            DATASET_REPO_ID,
+            SENT_PATH,
+            updated_sent_csv,
+            commit_message=f"Send keys for order {order_number} at {_utc_now_iso()}",
+        )
+        _upload_csv(
+            DATASET_REPO_ID,
+            KEYS_PATH,
+            updated_keys_csv,
+            commit_message=f"Mark keys as sent for order {order_number} at {_utc_now_iso()}",
+        )
+
+        # refresh cache immediately
+        _cache["ts"] = 0.0
+
+        # Send email
+        try:
+            send_key_email(order["email_address"], keys_str.split(","), order["contact_person"])
+            return keys_str, f"Keys sent successfully to {order['email_address']}."
+        except Exception as e:
+            return keys_str, f"Keys allocated: {keys_str}. Email failed: {str(e)}"
+
+# -------------------------
+# UI
+# -------------------------
+with gr.Blocks() as demo:
+    gr.Markdown(
+        "## Gradium Key Distribution System\n\n"
+        "Enter an **order number** to send keys to the customer.\n\n"
+    )
+
+    order_in = gr.Textbox(label="Order number", placeholder="e.g. 0000-1111")
+    send_btn = gr.Button("Send Keys", variant="primary")
+    keys_out = gr.Textbox(label="Keys sent", interactive=False)
+    status_out = gr.Textbox(label="Status", interactive=False)
+
+    send_btn.click(
+        fn=send_keys_for_order,
+        inputs=[order_in],
+        outputs=[keys_out, status_out],
+        api_name="send_keys",
+    )
+
+demo.queue()
+demo.launch()