@clawhub-zhangpei03-41a904c892
Use this skill whenever the user asks to analyze, verify, debug, reverse-engineer, or automate Kuaishou Genius「预算/预测/实际」页面 data flow (especially management-y...
---
name: kuaishou-genius-actual
description: Use this skill whenever the user asks to analyze, verify, debug, reverse-engineer, or automate Kuaishou Genius「预算/预测/实际」页面 data flow (especially management-yearly/actual). Trigger on requests about Network 抓包, API mapping, endpoint calls, auth/session reuse, payload reconstruction, or script-based API probing for genius.corp.kuaishou.com/budget-portal.
---
# Kuaishou Genius Actual API Skill
## Overview
This skill helps an agent quickly move from **Genius 页面操作** to **可复用的接口调用地图与脚本化验证** for the `management-yearly/actual` workflow.
Use it when the goal is to identify core backend endpoints, validate request dependencies, and build repeatable checks for Genius Actual data retrieval.
## Quick Start
1. Ensure login/session is valid for `genius.corp.kuaishou.com`.
2. Capture network around page reload and key filter actions.
3. Focus on `/budget-portal/api/*` requests; ignore static assets and telemetry unless debugging auth/risk.
4. Run script-based endpoint probe/client:
```bash
cd scripts
bash genius_api_probe.sh \
--base-url "https://genius.corp.kuaishou.com" \
--cookie "accessproxy_session=<YOUR_COOKIE>" \
--year 2026
python3 genius_client.py \
--cookie "accessproxy_session=<YOUR_COOKIE>" \
workflow --year 2026
```
5. Output a concise report with:
- reachable endpoints
- required params/payload hints
- dependency order
- known blockers/limitations
## Supported Capabilities
1. **Core API extraction**
- Identify actual business endpoints used by `management-yearly/actual`.
2. **API map generation**
- Build endpoint catalog: method, path, purpose, required params/body.
3. **Workflow reconstruction**
- Reconstruct request order from page load to ledger detail fetch.
4. **Scripted probing**
- Use `scripts/genius_api_probe.sh` to quickly verify endpoint reachability and baseline responses.
5. **Troubleshooting focus**
- Distinguish business API failures from:
- SSO/session expiration
- fingerprint/risk controls
- telemetry noise
## API Map (Core Business)
Base domain:
- `https://genius.corp.kuaishou.com`
Core endpoints observed in Actual flow:
- `GET /budget-portal/api/authority/user`
- Purpose: fetch user auth context.
- `GET /budget-portal/api/authority/org/tree`
- Purpose: org tree for selectors/permissions scope.
- `GET /budget-portal/api/horse-race-lamp/query?tabCode=management-yearly%2Factual`
- Purpose: tab-level notification/meta.
- `GET /budget-portal/api/description/act-latest-update-date`
- Purpose: latest actual update metadata.
- `GET /budget-portal/api/annual-actual/versions?year=<YEAR>`
- Purpose: available versions for selected year.
- `POST /budget-portal/api/actual-ledger/detail`
- Purpose: ledger detail dataset.
- Notes: requires JSON body shaped by current filters.
- `POST /budget-portal/api/actual-ledger/products`
- Purpose: product/metric dimension data for current view.
- Notes: requires JSON body shaped by current filters.
Non-core but commonly seen (usually ignore unless diagnosing):
- `log-sdk.ksapisrv.com/*` telemetry
- `mobile-device-info.corp.kuaishou.com/*` device/risk
- `h5-fingerprint.corp.kuaishou.com/*` fingerprint
## Workflow
### 1) Session check
- Confirm not redirected to SSO login.
- Verify `accessproxy_session` works for `genius.corp.kuaishou.com`.
### 2) Capture
- Reload target page:
- `https://genius.corp.kuaishou.com/management-yearly/actual`
- Capture all XHR/fetch.
### 3) Filter to business APIs
- Keep only `/budget-portal/api/` requests.
- Group by: authority → metadata → versions → ledger POSTs.
### 4) Rebuild minimal call chain
- Start with GET chain (auth/org/version).
- Then reproduce POST ledger calls with realistic payload.
### 5) Validate by script
- Run `genius_api_probe.sh` with cookie + year.
- Record HTTP code + brief body snippet.
### 6) Report
Always output:
- API list (method/path/purpose)
- call order
- required parameters/body fields (known/unknown)
- current blockers and next action
## Script Usage
Script paths:
- `scripts/genius_api_probe.sh`
- `scripts/genius_client.py`
What they do:
- `genius_api_probe.sh`: probes key GET APIs and sends placeholder POSTs for quick triage
- `genius_client.py`: structured client for core APIs (single endpoint or full workflow), supports custom JSON payload files
Required inputs:
- `--base-url` (default `https://genius.corp.kuaishou.com`)
- `--cookie` (must include valid `accessproxy_session=...`)
Optional:
- `--year` (default `2026`)
## Known Limitations
1. **SSO/session coupling**
- Without valid session cookie, requests fall back to SSO and API probing is invalid.
2. **Risk/fingerprint controls**
- Some environments may require device/fingerprint side requests; replay outside browser may fail.
3. **POST body incompleteness**
- `actual-ledger/detail` and `actual-ledger/products` need accurate business payload fields from live capture.
4. **Environment drift**
- static bundle versions and backend schema may change; always re-capture when results diverge.
5. **Permission scope**
- org tree and ledger visibility depend on account permissions; data differences are expected across users.
FILE:scripts/detail.json
{"dateAccuracy":"month","orgCodes":["E01"],"showAccuracy":"one","showZero":false,"transactionType":"nonInternal","year":"2025","excludeJuli":false,"versionCode":"Act","scenarioCode":"Act_Manual","actualYear":"2025"}
FILE:scripts/detail_2026_group.json
{"dateAccuracy":"month","orgCodes":["E01"],"showAccuracy":"one","showZero":false,"transactionType":"nonInternal","year":"2026","excludeJuli":false,"versionCode":"Act","scenarioCode":"Act_Manual","actualYear":"2026"}
FILE:scripts/genius_api_probe.sh
#!/usr/bin/env bash
set -euo pipefail
BASE_URL="https://genius.corp.kuaishou.com"
COOKIE=""
YEAR="2026"
while [[ $# -gt 0 ]]; do
case "$1" in
--base-url) BASE_URL="$2"; shift 2 ;;
--cookie) COOKIE="$2"; shift 2 ;;
--year) YEAR="$2"; shift 2 ;;
*) echo "Unknown arg: $1"; exit 1 ;;
esac
done
if [[ -z "$COOKIE" ]]; then
echo "ERROR: --cookie is required, e.g. --cookie 'accessproxy_session=xxx'"
exit 1
fi
function hit_get() {
local path="$1"
echo "\n>>> GET $path"
curl -sS -i \
-H "Cookie: $COOKIE" \
"$BASE_URL$path" | head -c 800
echo
}
function hit_post() {
local path="$1"
local body="$2"
echo "\n>>> POST $path"
curl -sS -i \
-H "Cookie: $COOKIE" \
-H "Content-Type: application/json" \
-X POST \
-d "$body" \
"$BASE_URL$path" | head -c 800
echo
}
hit_get "/budget-portal/api/authority/user"
hit_get "/budget-portal/api/authority/org/tree"
hit_get "/budget-portal/api/horse-race-lamp/query?tabCode=management-yearly%2Factual"
hit_get "/budget-portal/api/description/act-latest-update-date"
hit_get "/budget-portal/api/annual-actual/versions?year=YEAR"
# Placeholder payloads: adjust from live capture when needed.
hit_post "/budget-portal/api/actual-ledger/detail" '{"year":'"$YEAR"',"periodType":"MONTH"}'
hit_post "/budget-portal/api/actual-ledger/products" '{"year":'"$YEAR"',"periodType":"MONTH"}'
echo "\nDone. If POST returns validation error, capture real request body from browser Network and retry."
FILE:scripts/genius_client.py
#!/usr/bin/env python3
"""
Genius API client for core business endpoints in management-yearly/actual flow.
Usage examples:
python3 genius_client.py --cookie "accessproxy_session=xxx" user
python3 genius_client.py --cookie "accessproxy_session=xxx" versions --year 2026
python3 genius_client.py --cookie "accessproxy_session=xxx" detail --payload-file detail.json
python3 genius_client.py --cookie "accessproxy_session=xxx" workflow --year 2026
"""
from __future__ import annotations
import argparse
import json
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class ApiResponse:
ok: bool
status: int
url: str
data: Any
error: Optional[str] = None
class GeniusClient:
def __init__(self, base_url: str, cookie: str, timeout: int = 20, insecure: bool = False):
self.base_url = base_url.rstrip("/")
self.cookie = cookie
self.timeout = timeout
self._ssl_ctx = None
if insecure:
self._ssl_ctx = ssl.create_default_context()
self._ssl_ctx.check_hostname = False
self._ssl_ctx.verify_mode = ssl.CERT_NONE
def _request(self, method: str, path: str, *, query: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None) -> ApiResponse:
if not path.startswith("/"):
path = "/" + path
url = self.base_url + path
if query:
url += "?" + urllib.parse.urlencode(query, doseq=True)
headers = {
"Cookie": self.cookie,
"Accept": "application/json, text/plain, */*",
"User-Agent": "genius-client/1.0",
}
data: Optional[bytes] = None
if body is not None:
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url=url, method=method.upper(), headers=headers, data=data)
try:
with urllib.request.urlopen(req, timeout=self.timeout, context=self._ssl_ctx) as resp:
raw = resp.read()
content_type = resp.headers.get("Content-Type", "")
if "application/json" in content_type:
parsed = json.loads(raw.decode("utf-8", errors="replace"))
else:
parsed = raw.decode("utf-8", errors="replace")
return ApiResponse(True, resp.status, url, parsed)
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8", errors="replace") if e.fp else ""
return ApiResponse(False, e.code, url, raw, f"HTTPError {e.code}")
except Exception as e:
return ApiResponse(False, -1, url, "", f"{type(e).__name__}: {e}")
# ----- Core APIs -----
def get_user(self) -> ApiResponse:
return self._request("GET", "/budget-portal/api/authority/user")
def get_org_tree(self) -> ApiResponse:
return self._request("GET", "/budget-portal/api/authority/org/tree")
def get_horse_race(self, tab_code: str = "management-yearly/actual") -> ApiResponse:
return self._request("GET", "/budget-portal/api/horse-race-lamp/query", query={"tabCode": tab_code})
def get_latest_update_date(self) -> ApiResponse:
return self._request("GET", "/budget-portal/api/description/act-latest-update-date")
def get_versions(self, year: int) -> ApiResponse:
return self._request("GET", "/budget-portal/api/annual-actual/versions", query={"year": year})
def post_ledger_detail(self, payload: Dict[str, Any]) -> ApiResponse:
return self._request("POST", "/budget-portal/api/actual-ledger/detail", body=payload)
def post_ledger_products(self, payload: Dict[str, Any]) -> ApiResponse:
return self._request("POST", "/budget-portal/api/actual-ledger/products", body=payload)
DEFAULT_DETAIL_PAYLOAD = {
"year": 2026,
"periodType": "MONTH",
}
def load_payload(path: Optional[str], fallback: Dict[str, Any], year: int) -> Dict[str, Any]:
if path:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
payload = dict(fallback)
payload["year"] = year
return payload
def print_response(resp: ApiResponse, pretty: bool = True) -> None:
print(f"\n[{ 'OK' if resp.ok else 'ERR' }] {resp.status} {resp.url}")
if resp.error:
print(f"error: {resp.error}")
if isinstance(resp.data, (dict, list)):
if pretty:
print(json.dumps(resp.data, ensure_ascii=False, indent=2))
else:
print(json.dumps(resp.data, ensure_ascii=False))
else:
txt = str(resp.data)
print(txt[:2000] + ("..." if len(txt) > 2000 else ""))
def run_workflow(client: GeniusClient, year: int, detail_payload: Dict[str, Any], products_payload: Dict[str, Any], pretty: bool) -> int:
calls = [
("user", client.get_user),
("org_tree", client.get_org_tree),
("horse_race", client.get_horse_race),
("latest_update", client.get_latest_update_date),
("versions", lambda: client.get_versions(year)),
("detail", lambda: client.post_ledger_detail(detail_payload)),
("products", lambda: client.post_ledger_products(products_payload)),
]
failed = 0
for _, fn in calls:
resp = fn()
print_response(resp, pretty=pretty)
if not resp.ok:
failed += 1
return failed
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="Genius core API client")
p.add_argument("command", choices=["user", "org-tree", "horse-race", "latest-update", "versions", "detail", "products", "workflow"], help="API command")
p.add_argument("--base-url", default="https://genius.corp.kuaishou.com", help="Base URL")
p.add_argument("--cookie", required=True, help="Cookie string, e.g. 'accessproxy_session=xxx; other=yyy'")
p.add_argument("--year", type=int, default=2026, help="Year parameter")
p.add_argument("--payload-file", help="JSON payload file for detail/products")
p.add_argument("--products-payload-file", help="JSON payload file for products (workflow only)")
p.add_argument("--timeout", type=int, default=20, help="HTTP timeout seconds")
p.add_argument("--insecure", action="store_true", help="Disable TLS verification")
p.add_argument("--compact", action="store_true", help="Compact JSON output")
return p
def main() -> int:
args = build_parser().parse_args()
client = GeniusClient(
base_url=args.base_url,
cookie=args.cookie,
timeout=args.timeout,
insecure=args.insecure,
)
pretty = not args.compact
if args.command == "user":
print_response(client.get_user(), pretty=pretty)
elif args.command == "org-tree":
print_response(client.get_org_tree(), pretty=pretty)
elif args.command == "horse-race":
print_response(client.get_horse_race(), pretty=pretty)
elif args.command == "latest-update":
print_response(client.get_latest_update_date(), pretty=pretty)
elif args.command == "versions":
print_response(client.get_versions(args.year), pretty=pretty)
elif args.command == "detail":
payload = load_payload(args.payload_file, DEFAULT_DETAIL_PAYLOAD, args.year)
print_response(client.post_ledger_detail(payload), pretty=pretty)
elif args.command == "products":
payload = load_payload(args.payload_file, DEFAULT_DETAIL_PAYLOAD, args.year)
print_response(client.post_ledger_products(payload), pretty=pretty)
elif args.command == "workflow":
detail_payload = load_payload(args.payload_file, DEFAULT_DETAIL_PAYLOAD, args.year)
products_payload = load_payload(args.products_payload_file, detail_payload, args.year)
failed = run_workflow(client, args.year, detail_payload, products_payload, pretty=pretty)
if failed:
print(f"\nworkflow finished with {failed} failed call(s)")
return 2
print("\nworkflow finished successfully")
else:
raise ValueError("Unsupported command")
return 0
if __name__ == "__main__":
sys.exit(main())
FILE:scripts/products.json
{"dateAccuracy":"month","orgCodes":["E01"],"showAccuracy":"one","showZero":false,"transactionType":"nonInternal","year":"2025","versionCode":"Act","scenarioCode":"Act_Manual","actualYear":"2025"}