# pal_v1.py
#
# PAL v1
# Two commands:
# 1) ingest -> store external data
# 2) analyze -> analyze stored data
#
# Examples:
# python pal_v1.py ingest "{\"entity\":\"truck_17\",\"event_type\":\"shipment\",\"location\":\"taipei\",\"status\":\"delayed\",\"note\":\"flat tire\"}"
# python pal_v1.py analyze
#
# Requirements:
# pip install openai
# OPENAI_API_KEY in .env or environment
import json
import os
import sys
from pathlib import Path
from datetime import datetime, timezone
from typing import Any, Dict, List
from openai import OpenAI
# --------------------------------------------------
# 1.0 ENV / API KEY
# --------------------------------------------------
def load_dotenv(dotenv_path: str = ".env") -> None:
path = Path(dotenv_path)
if not path.exists():
return
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line or line.startswith("REM "):
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
os.environ.setdefault(key, value)
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY missing. Put it in .env or environment.")
client = OpenAI(api_key=api_key)
# --------------------------------------------------
# 1.1 FILES / CONSTANTS
# --------------------------------------------------
EVENTS_FILE = Path("pal_events.json")
ALLOWED_TOP_KEYS = {
"timestamp",
"entity",
"event_type",
"location",
"status",
"note",
}
REQUIRED_KEYS = {
"entity",
"event_type",
"location",
"status",
"note",
}
ANALYSIS_SCHEMA_TEXT = """
Return valid JSON only, with this exact top-level structure:
{
"summary": "short text summary",
"abnormal_events": [
{
"entity": "string",
"event_type": "string",
"location": "string",
"status": "string",
"reason": "string"
}
],
"problem_entities": ["string"],
"problem_locations": ["string"]
}
Rules:
- Return valid JSON only.
- Do not include markdown.
- "abnormal_events" should contain events that look problematic, unusual, delayed, failed, blocked, missing, or suspicious.
- "problem_entities" should list repeated or notable problematic entities.
- "problem_locations" should list repeated or notable problematic locations.
- If there are no abnormal events, return an empty list.
"""
# --------------------------------------------------
# 1.2 HELPERS
# --------------------------------------------------
def utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def load_events() -> List[Dict[str, Any]]:
if not EVENTS_FILE.exists():
return []
try:
data = json.loads(EVENTS_FILE.read_text(encoding="utf-8"))
if not isinstance(data, list):
raise ValueError("Events file must contain a JSON list.")
return data
except Exception as e:
raise RuntimeError(f"Failed to load {EVENTS_FILE}: {e}")
def save_events(events: List[Dict[str, Any]]) -> None:
EVENTS_FILE.write_text(
json.dumps(events, indent=2, ensure_ascii=False),
encoding="utf-8"
)
def validate_event(event: Dict[str, Any]) -> List[str]:
errors: List[str] = []
if not isinstance(event, dict):
return ["Event must be a JSON object."]
for key in REQUIRED_KEYS:
if key not in event:
errors.append(f"Missing required key: '{key}'.")
for key in event.keys():
if key not in ALLOWED_TOP_KEYS:
errors.append(f"Unexpected key: '{key}'.")
for key in REQUIRED_KEYS:
if key in event and not isinstance(event[key], str):
errors.append(f"'{key}' must be a string.")
if "timestamp" in event and not isinstance(event["timestamp"], str):
errors.append("'timestamp' must be a string.")
return errors
def normalize_event(event: Dict[str, Any]) -> Dict[str, Any]:
out = dict(event)
if "timestamp" not in out:
out["timestamp"] = utc_now_iso()
return out
def print_usage() -> None:
print(
"Usage:\n"
" python pal_v1.py ingest '<json_event>'\n"
" python pal_v1.py analyze\n\n"
"Example:\n"
' python pal_v1.py ingest "{\\"entity\\":\\"truck_17\\",\\"event_type\\":\\"shipment\\",\\"location\\":\\"taipei\\",\\"status\\":\\"delayed\\",\\"note\\":\\"flat tire\\"}"\n'
" python pal_v1.py analyze"
)
# --------------------------------------------------
# 1.3 COMMAND: INGEST
# --------------------------------------------------
def cmd_ingest(event_json_text: str) -> None:
try:
event = json.loads(event_json_text)
except Exception as e:
print("INGEST FAILED")
print(f"Invalid JSON input: {e}")
return
errors = validate_event(event)
if errors:
print("INGEST FAILED")
for err in errors:
print(f"- {err}")
return
event = normalize_event(event)
events = load_events()
events.append(event)
save_events(events)
print("INGEST OK")
print(f"Saved to: {EVENTS_FILE.resolve()}")
print("Event:")
print(json.dumps(event, indent=2, ensure_ascii=False))
# --------------------------------------------------
# 1.4 COMMAND: ANALYZE
# --------------------------------------------------
def build_analysis_messages(events: List[Dict[str, Any]]) -> List[Dict[str, str]]:
events_json = json.dumps(events, indent=2, ensure_ascii=False)
return [
{
"role": "system",
"content": (
"You are a structured data analysis model.\n\n"
f"{ANALYSIS_SCHEMA_TEXT}"
),
},
{
"role": "user",
"content": (
"Analyze the following stored events.\n\n"
f"{events_json}"
),
},
]
def request_analysis(events: List[Dict[str, Any]]) -> Dict[str, Any]:
messages = build_analysis_messages(events)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
response_format={"type": "json_object"},
)
content = response.choices[0].message.content
return json.loads(content)
def cmd_analyze() -> None:
events = load_events()
if not events:
print("ANALYZE FAILED")
print("No events stored yet.")
return
print("=== STORED EVENTS ===")
print(json.dumps(events, indent=2, ensure_ascii=False))
try:
analysis = request_analysis(events)
except Exception as e:
print("ANALYZE FAILED")
print(str(e))
return
print("\n=== ANALYSIS ===")
print(json.dumps(analysis, indent=2, ensure_ascii=False))
# --------------------------------------------------
# 1.5 MAIN
# --------------------------------------------------
def main() -> None:
if len(sys.argv) < 2:
print_usage()
return
command = sys.argv[1].strip().lower()
if command == "ingest":
if len(sys.argv) < 3:
print("Missing JSON event for ingest.\n")
print_usage()
return
event_json_text = sys.argv[2]
cmd_ingest(event_json_text)
elif command == "analyze":
cmd_analyze()
else:
print(f"Unknown command: {command}\n")
print_usage()
if __name__ == "__main__":
main()