Copilot のコード実行── Copilot がコードを書いて実行する場面と、その確かめ方

from pathlib import Path

# Limits from the Copilot Studio docs (structured data in agents, preview)
AGENT_MAX_MB = 16
AGENT_MAX_FILES = 10
TESTED_TYPES = {".csv", ".xlsx"}   # types in Microsoft's extensively tested scenarios


def precheck(paths, mode="agent"):
    paths = [Path(p) for p in paths]
    problems = []
    if mode == "prompt":
        # prompts: one uploaded file per run; no size limit is documented
        if len(paths) > 1:
            problems.append(f"prompt takes one file, got {len(paths)}")
        return problems
    if len(paths) > AGENT_MAX_FILES:
        problems.append(f"too many files: {len(paths)} > {AGENT_MAX_FILES}")
    for p in paths:
        size_mb = p.stat().st_size / (1024 * 1024)
        if size_mb > AGENT_MAX_MB:
            problems.append(f"{p.name}: {size_mb:.1f} MB > {AGENT_MAX_MB} MB")
        if p.suffix.lower() not in TESTED_TYPES:
            problems.append(f"{p.name}: {p.suffix} is outside the tested CSV/Excel scenarios")
    return problems


files = ["samples/ledger.csv", "samples/request_ledger.xlsx", "samples/doc_a.docx"]
for mode in ("agent", "prompt"):
    issues = precheck(files, mode)
    print(mode, "OK" if not issues else "issues:")
    for i in issues:
        print("  -", i)

import os
import time

import requests
from azure.identity import InteractiveBrowserCredential

ENV_URL = os.environ.get("DATAVERSE_URL", "https://example.crm.dynamics.com")
MODEL_ID = os.environ.get("AI_MODEL_ID", "")   # msdyn_aimodelid of the prompt
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
API = f"{ENV_URL}/api/data/v9.2"
HEADERS = {"Accept": "application/json", "OData-MaxVersion": "4.0", "OData-Version": "4.0"}


def predict_body(inputs):
    # Prompt inputs travel in requestv2 as an open type; keys must match the prompt's input names
    return {"version": "2.0",
            "requestv2": {"@odata.type": "#Microsoft.Dynamics.CRM.expando", **inputs}}


def wait_seconds(value, default=5):
    # overrideRetryAfter format is not documented; fall back to a fixed wait
    try:
        return max(1, int(float(value)))
    except (TypeError, ValueError):
        return default


def output_of(resp):
    v2 = resp["responsev2"]
    out = v2["predictionOutput"]
    return {"status": v2.get("operationStatus"), "text": out.get("text", ""),
            "code": out.get("code", ""), "logs": out.get("logs", ""),
            "files": [f.get("file_name") for f in out.get("files", [])]}


def main():
    cred = InteractiveBrowserCredential(tenant_id=os.environ["TENANT_ID"], client_id=os.environ["CLIENT_ID"])
    token = cred.get_token(f"{ENV_URL}/user_impersonation").token
    session = requests.Session()
    session.headers.update({**HEADERS, "Authorization": f"Bearer {token}"})
    url = f"{API}/msdyn_aimodels({MODEL_ID})/Microsoft.Dynamics.CRM.Predict"
    body = predict_body({"question": "Count requests by document type"})
    if DRY_RUN:
        print("DRY_RUN: would POST", url, body)
        return
    r = session.post(url, json=body, timeout=120)
    r.raise_for_status()
    data = r.json()
    for _ in range(30):
        if str(data.get("overrideHttpStatusCode")) != "202":
            break
        time.sleep(wait_seconds(data.get("overrideRetryAfter")))
        data = session.get(data["overrideLocation"], timeout=60).json()
    result = output_of(data)
    print(result["status"], result["files"])
    print(result["code"])


if __name__ == "__main__":
    main()

import base64
import json
from pathlib import Path

SRC = "samples/excel_predict_response.json"   # a saved Predict response
OUT = Path("out_predict")
OUT.mkdir(exist_ok=True)

resp = json.loads(Path(SRC).read_text(encoding="utf-8"))
v2 = resp["responsev2"]
if v2.get("operationStatus") != "Success":
    raise SystemExit(f"prediction did not succeed: {v2.get('operationStatus')}")
out = v2["predictionOutput"]

code = out.get("code", "")
(OUT / "executed_code.py").write_text(code, encoding="utf-8")
(OUT / "logs.txt").write_text(out.get("logs", ""), encoding="utf-8")
for f in out.get("files", []):
    name = Path(f["file_name"]).name   # keep the base name only
    (OUT / name).write_bytes(base64.b64decode(f["base64_content"]))
    print("file:", name, f.get("content_type"))

print("code lines:", len(code.splitlines()), "| finish:", out.get("finishReason"))
# Lines that change the row count deserve a closer look
WATCH = ("dropna", "fillna", "drop_duplicates", ".query(")
for no, line in enumerate(code.splitlines(), 1):
    if any(w in line for w in WATCH):
        print(f"review line {no}: {line.strip()}")

import pandas as pd

SRC = "samples/request_ledger.xlsx"
# Numbers copied from Copilot's answer to "average findings by department" (example)
copilot = {"営業部": 4.6, "広報部": 4.4, "人事部": 4.1}

df = pd.read_excel(SRC)
mine = df.groupby("依頼部署")["指摘件数"].mean().round(1)

check = pd.DataFrame({"copilot": pd.Series(copilot, dtype=float), "pandas": mine})
check["diff"] = (check["copilot"] - check["pandas"]).round(2)
print(check.to_string())

mismatch = check[(check["diff"].abs() > 0.05) | check.isna().any(axis=1)]
print("rows used:", len(df), "| mismatch:", list(mismatch.index))

import pandas as pd   # pre-imported in Python in Excel; explicit here

BOOK = "samples/request_ledger.xlsx"
# Map each reference used in the copied code to a sheet in the saved book
REFS = {"RequestLedger[#All]": "依頼台帳"}


def xl(ref, headers=False):
    # Local stand-in for Python in Excel's xl(): reads saved cell values only
    return pd.read_excel(BOOK, sheet_name=REFS[ref], header=0 if headers else None)


# ---- code copied from the Python cell (unchanged) ----
df = xl("RequestLedger[#All]", headers=True)
df["依頼月"] = pd.to_datetime(df["依頼日"]).dt.to_period("M")
monthly = df.pivot_table(index="依頼月", columns="状態", values="依頼ID",
                         aggfunc="count", fill_value=0)
# ---- end of copied code ----

print(monthly.to_string())
print("rows read:", len(df))

from openpyxl import load_workbook

BEFORE = "samples/request_ledger.xlsx"      # copy saved before the Copilot edit
AFTER = "samples/excel_ledger_after.xlsx"   # workbook after the edit


def cells(path):
    wb = load_workbook(path)   # default mode: formulas as text, nothing recalculated
    found = {}
    for ws in wb.worksheets:
        for row in ws.iter_rows():
            for c in row:
                if c.value is not None:
                    found[(ws.title, c.coordinate)] = c.value
    return found, wb.sheetnames


a, sheets_a = cells(BEFORE)
b, sheets_b = cells(AFTER)
print("added sheets:", [s for s in sheets_b if s not in sheets_a])
print("removed sheets:", [s for s in sheets_a if s not in sheets_b])

changed = sorted(k for k in a.keys() & b.keys() if a[k] != b[k])
new_in_old_sheets = sorted(k for k in b.keys() - a.keys() if k[0] in sheets_a)
cleared = sorted(a.keys() - b.keys())
for k in changed:
    print("changed", k, repr(a[k]), "->", repr(b[k]))
print(f"changed={len(changed)} new_in_existing_sheets={len(new_in_old_sheets)} cleared={len(cleared)}")