Gen Paylink Govilo

[project]
name = "govilo-to-go"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["requests"]

[project.scripts]
create-link = "scripts.workflow_create:main"

[project.optional-dependencies]
dev = ["pytest"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["scripts"]

[tool.pytest.ini_options]
testpaths = ["tests"]

from pathlib import Path

import requests


class ApiError(Exception):
    def __init__(self, message: str, code: int | None = None):
        super().__init__(message)
        self.code = code


class GoviloClient:
    def __init__(self, api_key: str, base_url: str):
        self._base_url = base_url.rstrip("/")
        self._headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def _check_response(self, resp: requests.Response) -> dict:
        body = resp.json()
        if body.get("code", 0) != 0:
            raise ApiError(body.get("msg", "unknown error"), body.get("code"))
        return body["data"]

    def presign(self, seller_address: str) -> dict:
        resp = requests.post(
            f"{self._base_url}/api/v1/bot/uploads/presign",
            headers=self._headers,
            json={"seller_address": seller_address},
        )
        return self._check_response(resp)

    def upload(self, upload_url: str, zip_path: Path) -> None:
        with open(zip_path, "rb") as f:
            resp = requests.put(
                upload_url,
                headers={"Content-Type": "application/zip"},
                data=f,
            )
        if resp.status_code != 200:
            raise ApiError(f"Upload failed: HTTP {resp.status_code}")

    def create_item(
        self,
        session_id: str,
        title: str,
        price: str,
        description: str = "",
    ) -> dict:
        resp = requests.post(
            f"{self._base_url}/api/v1/bot/items",
            headers=self._headers,
            json={
                "session_id": session_id,
                "title": title,
                "price": price,
                "description": description,
            },
        )
        return self._check_response(resp)

from dataclasses import dataclass
import os


BASE_URL = "https://api.unlock.govilo.xyz"


class ConfigError(Exception):
    pass


@dataclass(frozen=True)
class Config:
    api_key: str
    seller_address: str
    base_url: str = BASE_URL


def load_config(
    cli_api_key: str | None = None,
    cli_address: str | None = None,
) -> Config:
    api_key = cli_api_key or os.environ.get("GOVILO_API_KEY")
    if not api_key:
        raise ConfigError(
            "API Key not configured. See references/setup-guide.md for registration steps. "
            "Register at https://govilo.xyz/ then set GOVILO_API_KEY=sk_live_xxx"
        )

    seller_address = cli_address or os.environ.get("SELLER_ADDRESS")
    if not seller_address:
        raise ConfigError(
            "Seller address required (Base chain). Use --address 0x... or set SELLER_ADDRESS env var. "
            "See references/setup-guide.md for wallet setup"
        )

    return Config(api_key=api_key, seller_address=seller_address)

import tempfile
import zipfile
from pathlib import Path

MAX_ZIP_SIZE = 20 * 1024 * 1024  # 20 MB
MAX_FILE_COUNT = 20


class PackageError(Exception):
    pass


def _validate_zip(zp: Path) -> None:
    if zp.stat().st_size > MAX_ZIP_SIZE:
        raise PackageError(f"ZIP file exceeds 20 MB limit")
    with zipfile.ZipFile(zp) as zf:
        if len(zf.namelist()) > MAX_FILE_COUNT:
            raise PackageError(f"ZIP contains more than {MAX_FILE_COUNT} files")


def _zip_paths(paths: list[Path], dest: Path) -> None:
    with zipfile.ZipFile(dest, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in paths:
            zf.write(p, p.name)


def package(inputs: list[str]) -> Path:
    paths = [Path(p) for p in inputs]

    for p in paths:
        if not p.exists():
            raise PackageError(f"Path not found: {p}")

    # Single .zip file — passthrough
    if len(paths) == 1 and paths[0].suffix == ".zip":
        _validate_zip(paths[0])
        return paths[0]

    # Single directory — zip its contents
    if len(paths) == 1 and paths[0].is_dir():
        files = [f for f in paths[0].rglob("*") if f.is_file()]
        if len(files) > MAX_FILE_COUNT:
            raise PackageError(f"Directory contains more than {MAX_FILE_COUNT} files")
        dest = Path(tempfile.mktemp(suffix=".zip"))
        with zipfile.ZipFile(dest, "w", zipfile.ZIP_DEFLATED) as zf:
            for f in files:
                zf.write(f, f.relative_to(paths[0]))
        _validate_zip(dest)
        return dest

    # Multiple files — zip them together
    if len(paths) > MAX_FILE_COUNT:
        raise PackageError(f"Input exceeds {MAX_FILE_COUNT} files limit")
    dest = Path(tempfile.mktemp(suffix=".zip"))
    _zip_paths(paths, dest)
    _validate_zip(dest)
    return dest

import argparse
import json
import sys
from pathlib import Path

from scripts.api_client import GoviloClient, ApiError
from scripts.config import load_config, ConfigError
from scripts.packager import package, PackageError


def _output(data: dict) -> None:
    print(json.dumps(data, ensure_ascii=False))


def _fail(error: str, code: int | None = None) -> None:
    out = {"ok": False, "error": error}
    if code is not None:
        out["code"] = code
    _output(out)
    sys.exit(1)


def main() -> None:
    parser = argparse.ArgumentParser(description="Upload files and create Govilo unlock link")
    parser.add_argument("--input", required=True, action="append", dest="inputs", help="ZIP, folder, or file path (repeatable)")
    parser.add_argument("--title", required=True, help="Product title")
    parser.add_argument("--price", required=True, help="Price in USDC (e.g. 5.00)")
    parser.add_argument("--address", default=None, help="Seller EVM wallet address (overrides SELLER_ADDRESS env)")
    parser.add_argument("--description", default="", help="Product description")
    args = parser.parse_args()

    try:
        cfg = load_config(cli_api_key=None, cli_address=args.address)
    except ConfigError as e:
        _fail(str(e))

    try:
        zip_path = package(args.inputs)
    except PackageError as e:
        _fail(str(e))

    is_temp = zip_path not in [Path(p) for p in args.inputs]
    try:
        client = GoviloClient(api_key=cfg.api_key, base_url=cfg.base_url)

        try:
            presign_data = client.presign(cfg.seller_address)
            client.upload(presign_data["upload_url"], zip_path)
            item_data = client.create_item(
                session_id=presign_data["session_id"],
                title=args.title,
                price=args.price,
                description=args.description,
            )
        except ApiError as e:
            _fail(str(e), code=e.code)

        _output({
            "ok": True,
            "unlock_url": item_data["unlock_url"],
            "item_id": item_data["id"],
            "file_count": item_data["file_count"],
            "total_size": item_data["total_size"],
        })
    finally:
        if is_temp:
            zip_path.unlink(missing_ok=True)