実行環境の選び方── Python in Excel・手元の venv・Azure Functions・ノートブック

"""Record the Python runtime and package versions next to your results (JSON Lines)."""
import datetime
import importlib.metadata as md
import json
import platform
import sys

PACKAGES = ["pandas", "openpyxl", "python-docx", "azure-functions", "microsoft-agents-hosting-core"]


def runtime_record(packages):
    versions = {}
    for name in packages:
        try:
            versions[name] = md.version(name)
        except md.PackageNotFoundError:
            versions[name] = None
    return {
        "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
        "python": platform.python_version(),
        "implementation": platform.python_implementation(),
        "os": platform.platform(),
        "executable": sys.executable,
        "in_venv": sys.prefix != sys.base_prefix,
        "packages": versions,
    }


if __name__ == "__main__":
    rec = runtime_record(PACKAGES)
    with open("out_runtime_log.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    print("python", rec["python"], "| venv:", rec["in_venv"])
    for name, version in rec["packages"].items():
        print(f"  {name}: {version or 'not installed'}")

"""Compare a pinned requirements file with the packages installed in the current environment."""
import importlib.metadata as md
import os
import re
import sys

REQ_FILE = os.environ.get("REQ_FILE", "samples/agents_requirements.txt")
PIN = re.compile(r"^([A-Za-z0-9._-]+)==([^\s;#]+)")


def read_pins(path):
    pins = {}
    with open(path, encoding="utf-8") as f:
        for line in f:
            m = PIN.match(line.strip())
            if m:
                pins[m.group(1)] = m.group(2)
    return pins


def compare(pins):
    rows = []
    for name, wanted in sorted(pins.items()):
        try:
            have = md.version(name)
        except md.PackageNotFoundError:
            have = None
        rows.append((name, wanted, have, "ok" if have == wanted else "MISMATCH"))
    return rows


def freeze_lines():
    """The same shape as `python -m pip freeze`, without starting another process."""
    return sorted(f"{d.metadata['Name']}=={d.version}" for d in md.distributions())


if __name__ == "__main__":
    rows = compare(read_pins(REQ_FILE))
    for name, wanted, have, state in rows:
        print(f"{name:<16} want {wanted:<8} have {str(have):<8} {state}")
    with open("out_freeze.txt", "w", encoding="utf-8") as f:
        f.write("\n".join(freeze_lines()) + "\n")
    bad = [r for r in rows if r[3] != "ok"]
    print("mismatches:", len(bad))
    sys.exit(1 if bad and os.environ.get("STRICT") == "1" else 0)

# =PY cell on the first sheet: show the runtime used for this workbook's Python results
import sys

import numpy as np
import pandas as pd

df = xl("ledger[#All]", headers=True)
pd.DataFrame({
    "item": ["python", "pandas", "numpy", "ledger rows"],
    "value": [sys.version.split()[0], pd.__version__, np.__version__, len(df)],
})

"""function_app.py: settings come from app settings in Azure and from local.settings.json on your machine."""
import csv
import json
import os

import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="open_requests", methods=["GET"])
def open_requests(req: func.HttpRequest) -> func.HttpResponse:
    open_states = set(os.getenv("OPEN_STATES", "受付,確認中").split(","))
    with open(os.getenv("LEDGER_PATH", "ledger.csv"), encoding="utf-8-sig", newline="") as f:
        rows = [r for r in csv.DictReader(f) if r["状態"] in open_states]
    dept = req.params.get("dept")
    if dept:
        rows = [r for r in rows if r["依頼部署"] == dept]
    body = {"count": len(rows), "ids": [r["依頼ID"] for r in rows]}
    return func.HttpResponse(json.dumps(body, ensure_ascii=False), mimetype="application/json")


def load_local_settings(path):
    """Copy the Values section into os.environ; values already set in the environment win."""
    with open(path, encoding="utf-8") as f:
        values = json.load(f).get("Values", {})
    for name, value in values.items():
        os.environ.setdefault(name, str(value))
    return sorted(values)


if __name__ == "__main__":
    # Quick check without Core Tools (azure-functions >= 1.21.0). Use `func start` for the real host.
    print("settings:", load_local_settings(os.getenv("LOCAL_SETTINGS", "samples/agents_local.settings.json")))
    req = func.HttpRequest(method="GET", url="/api/open_requests", body=b"", params={"dept": "広報部"})
    resp = open_requests.build().get_user_function()(req)
    print(resp.status_code, resp.get_body().decode("utf-8"))

"""Check host.json functionTimeout and the target Python version against hosting plan limits."""
import json
import os
import re

# Copied from "Azure Functions Scale and Hosting" and "Supported Languages in Azure Functions" on 2026-09-12.
PLAN_MAX_MINUTES = {"consumption": 10, "flex": None, "premium": None, "dedicated": None}  # None = unbounded
HTTP_RESPONSE_LIMIT_SECONDS = 230
PYTHON_GA = ["3.10", "3.11", "3.12", "3.13", "3.14"]
LINUX_CONSUMPTION_LAST = "3.12"
TIMESPAN = re.compile(r"^(?:(\d+)\.)?(\d{1,2}):(\d{2}):(\d{2})$")


def to_minutes(value):
    text = str(value).strip()
    if text == "-1":
        return None
    m = TIMESPAN.match(text)
    if not m:
        raise ValueError(f"not a [d.]hh:mm:ss timespan: {value!r}")
    d, h, mi, s = (int(x or 0) for x in m.groups())
    if h > 23 or mi > 59 or s > 59:
        raise ValueError(f"out of range: {value!r}")
    return d * 1440 + h * 60 + mi + s / 60


def version_key(v):
    return tuple(int(p) for p in v.split("."))


def check(host, plan, python_version, http_seconds):
    notes = []
    minutes = to_minutes(host.get("functionTimeout", "-1"))
    limit = PLAN_MAX_MINUTES[plan]
    if limit is not None and (minutes is None or minutes > limit):
        notes.append(f"functionTimeout {host.get('functionTimeout')} exceeds the {plan} maximum of {limit} min")
    if http_seconds > HTTP_RESPONSE_LIMIT_SECONDS:
        notes.append(f"HTTP work of {http_seconds}s exceeds {HTTP_RESPONSE_LIMIT_SECONDS}s: reply first, finish later")
    if python_version not in PYTHON_GA:
        notes.append(f"Python {python_version} is not in the supported list")
    elif plan == "consumption" and version_key(python_version) > version_key(LINUX_CONSUMPTION_LAST):
        notes.append(f"Linux Consumption stops at Python {LINUX_CONSUMPTION_LAST}")
    return notes


if __name__ == "__main__":
    with open(os.environ.get("HOST_JSON", "samples/agents_host.json"), encoding="utf-8") as f:
        host = json.load(f)
    print("functionTimeout:", host.get("functionTimeout"), "=", to_minutes(host.get("functionTimeout", "-1")), "min")
    for plan, py, seconds in [("consumption", "3.13", 300), ("flex", "3.13", 120)]:
        found = check(host, plan, py, seconds)
        print(f"[{plan} / Python {py} / HTTP {seconds}s]", "ok" if not found else "")
        for note in found:
            print("  -", note)

# First cell of a Fabric notebook: stop early if the session's libraries differ from the pinned set
import importlib.metadata as md
import json
import sys

EXPECTED = {"pandas": "3.0.5", "openpyxl": "3.1.5"}  # keep in sync with the environment (Full mode)


def library_report(expected):
    report = {"python": sys.version.split()[0], "libraries": {}, "mismatch": []}
    for name, wanted in expected.items():
        try:
            have = md.version(name)
        except md.PackageNotFoundError:
            have = None
        report["libraries"][name] = have
        if have != wanted:
            report["mismatch"].append(f"{name}: want {wanted}, have {have}")
    return report


report = library_report(EXPECTED)
print(json.dumps(report, ensure_ascii=False))
if report["mismatch"]:
    raise RuntimeError("library mismatch: " + "; ".join(report["mismatch"]))