(pal_events3.json... just a copy of pal_events2.json)
[
{
"entity": "truck_17",
"event_type": "shipment",
"location": "taipei",
"status": "delayed",
"note": "flat tire",
"timestamp": "2026-03-27T13:04:40.095404+00:00"
},
{
"entity": "truck_17",
"event_type": "shipment",
"location": "taipei",
"status": "delayed",
"note": "engine issue",
"timestamp": "2026-03-27T13:09:18.778733+00:00"
},
{
"entity": "truck_22",
"event_type": "shipment",
"location": "kaohsiung",
"status": "ok",
"note": "arrived on time",
"timestamp": "2026-03-27T13:09:27.448964+00:00"
},
{
"entity": "truck_31",
"event_type": "shipment",
"location": "tainan",
"status": "blocked",
"note": "road closure",
"timestamp": "2026-03-27T13:09:36.279725+00:00"
}
]
# pal_v3.py
# pal_v3.py
#
# PAL v3
# Commands:
# 1) ingest -> store external data
# 2) analyze -> analyze all stored data
# 3) query -> exact structured filter, then analyze matching events
# 4) ask -> natural language -> structured filter -> query -> analyze
#
# Bash examples:
# python pal_v3.py ingest '{"entity":"truck_17","event_type":"shipment","location":"taipei","status":"delayed","note":"flat tire"}'
# python pal_v3.py analyze
# python pal_v3.py query '{"entity":"truck_17"}'
# python pal_v3.py ask "show delayed events in taipei"
# python pal_v3.py ask "analyze truck_17"
# python pal_v3.py ask "show blocked shipments in tainan"
# python pal_v3.py ask "show all events"
#
# 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
# --------------------------------------------------
# 3.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)
# --------------------------------------------------
# 3.1 FILES / CONSTANTS
# --------------------------------------------------
EVENTS_FILE = Path("pal_events3.json")
ALLOWED_TOP_KEYS = {
"timestamp",
"entity",
"event_type",
"location",
"status",
"note",
}
REQUIRED_KEYS = {
"entity",
"event_type",
"location",
"status",
"note",
}
QUERYABLE_KEYS = {
"timestamp",
"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.
"""
FILTER_SCHEMA_TEXT = """
Return valid JSON only, with this exact top-level structure:
{
"mode": "all" OR "filter",
"filter": {}
}
Rules:
- Return valid JSON only.
- Do not include markdown.
- Allowed filter keys: timestamp, entity, event_type, location, status, note
- Filter values must be strings.
- Use mode = "all" only if the user clearly wants all events.
- Use mode = "filter" when the user is asking about a subset.
- The filter object may contain one or more key/value pairs.
- For "show delayed events in taipei", return:
{"mode":"filter","filter":{"status":"delayed","location":"taipei"}}
- For "analyze truck_17", return:
{"mode":"filter","filter":{"entity":"truck_17"}}
- For "show all events", return:
{"mode":"all","filter":{}}
"""
# --------------------------------------------------
# 3.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 validate_query_filter(query_filter: Dict[str, Any]) -> List[str]:
errors: List[str] = []
if not isinstance(query_filter, dict):
return ["Query filter must be a JSON object."]
if len(query_filter) == 0:
errors.append("Query filter must not be empty.")
for key, value in query_filter.items():
if key not in QUERYABLE_KEYS:
errors.append(f"Query key '{key}' is not allowed.")
if not isinstance(value, str):
errors.append(f"Query value for '{key}' must be a string.")
return errors
def event_matches_filter(event: Dict[str, Any], query_filter: Dict[str, str]) -> bool:
for key, wanted_value in query_filter.items():
actual_value = event.get(key)
if not isinstance(actual_value, str):
return False
if actual_value.lower() != wanted_value.lower():
return False
return True
def select_matching_events(
events: List[Dict[str, Any]],
query_filter: Dict[str, str]
) -> List[Dict[str, Any]]:
return [event for event in events if event_matches_filter(event, query_filter)]
def print_usage() -> None:
print(
"Usage:\n"
" python pal_v3.py ingest '<json_event>'\n"
" python pal_v3.py analyze\n"
" python pal_v3.py query '<json_filter>'\n"
" python pal_v3.py ask '<natural language query>'\n\n"
"Bash examples:\n"
' python pal_v3.py ingest \'{"entity":"truck_17","event_type":"shipment","location":"taipei","status":"delayed","note":"flat tire"}\'\n'
' python pal_v3.py query \'{"entity":"truck_17"}\'\n'
' python pal_v3.py ask "show delayed events in taipei"\n'
' python pal_v3.py ask "analyze truck_17"\n'
' python pal_v3.py ask "show all events"\n'
" python pal_v3.py analyze"
)
def print_events_block(title: str, events: List[Dict[str, Any]]) -> None:
print(title)
print(json.dumps(events, indent=2, ensure_ascii=False))
# --------------------------------------------------
# 3.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))
# --------------------------------------------------
# 3.4 ANALYSIS
# --------------------------------------------------
def build_analysis_messages(
events: List[Dict[str, Any]],
context_label: str
) -> 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": (
f"Analyze the following stored events.\n"
f"Context: {context_label}\n\n"
f"{events_json}"
),
},
]
def request_analysis(events: List[Dict[str, Any]], context_label: str) -> Dict[str, Any]:
messages = build_analysis_messages(events, context_label)
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)
# --------------------------------------------------
## 3.5 NATURAL LANGUAGE -> FILTER
# --------------------------------------------------
def build_filter_messages(user_query: str) -> List[Dict[str, str]]:
return [
{
"role": "system",
"content": (
"You convert natural language queries into structured event filters.\n\n"
f"{FILTER_SCHEMA_TEXT}"
),
},
{
"role": "user",
"content": user_query,
},
]
def request_structured_filter(user_query: str) -> Dict[str, Any]:
messages = build_filter_messages(user_query)
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 validate_filter_request(obj: Dict[str, Any]) -> List[str]:
errors: List[str] = []
if not isinstance(obj, dict):
return ["Filter request must be a JSON object."]
if "mode" not in obj:
errors.append("Missing required key: 'mode'.")
if "filter" not in obj:
errors.append("Missing required key: 'filter'.")
mode = obj.get("mode")
filter_obj = obj.get("filter")
if mode not in {"all", "filter"}:
errors.append("mode must be 'all' or 'filter'.")
if not isinstance(filter_obj, dict):
errors.append("filter must be a JSON object.")
else:
for key, value in filter_obj.items():
if key not in QUERYABLE_KEYS:
errors.append(f"Filter key '{key}' is not allowed.")
if not isinstance(value, str):
errors.append(f"Filter value for '{key}' must be a string.")
if mode == "filter" and isinstance(filter_obj, dict) and len(filter_obj) == 0:
errors.append("filter must not be empty when mode='filter'.")
if mode == "all" and isinstance(filter_obj, dict) and len(filter_obj) != 0:
errors.append("filter must be empty when mode='all'.")
return errors
# --------------------------------------------------
# 3.6 COMMAND: ANALYZE ALL
# --------------------------------------------------
def cmd_analyze() -> None:
events = load_events()
if not events:
print("ANALYZE FAILED")
print("No events stored yet.")
return
print_events_block("=== STORED EVENTS (ALL) ===", events)
try:
analysis = request_analysis(events, context_label="all events")
except Exception as e:
print("ANALYZE FAILED")
print(str(e))
return
print("\n=== ANALYSIS (ALL) ===")
print(json.dumps(analysis, indent=2, ensure_ascii=False))
# --------------------------------------------------
# 3.7 COMMAND: QUERY
# --------------------------------------------------
def run_query_filter(query_filter: Dict[str, str]) -> None:
errors = validate_query_filter(query_filter)
if errors:
print("QUERY FAILED")
for err in errors:
print(f"- {err}")
return
events = load_events()
if not events:
print("QUERY FAILED")
print("No events stored yet.")
return
matching_events = select_matching_events(events, query_filter)
print("=== QUERY FILTER ===")
print(json.dumps(query_filter, indent=2, ensure_ascii=False))
print(f"\n=== MATCH COUNT ===\n{len(matching_events)}")
if not matching_events:
print("\n=== MATCHING EVENTS ===")
print("[]")
return
print_events_block("\n=== MATCHING EVENTS ===", matching_events)
context_label = f"query filter = {json.dumps(query_filter, ensure_ascii=False)}"
try:
analysis = request_analysis(matching_events, context_label=context_label)
except Exception as e:
print("QUERY FAILED")
print(str(e))
return
print("\n=== ANALYSIS (MATCHING EVENTS ONLY) ===")
print(json.dumps(analysis, indent=2, ensure_ascii=False))
def cmd_query(query_json_text: str) -> None:
try:
query_filter = json.loads(query_json_text)
except Exception as e:
print("QUERY FAILED")
print(f"Invalid JSON filter: {e}")
return
run_query_filter(query_filter)
# --------------------------------------------------
## 3.8 COMMAND: ASK
# --------------------------------------------------
def cmd_ask(user_query: str) -> None:
events = load_events()
if not events:
print("ASK FAILED")
print("No events stored yet.")
return
print("=== NATURAL LANGUAGE QUERY ===")
print(user_query)
try:
filter_request = request_structured_filter(user_query)
except Exception as e:
print("ASK FAILED")
print(f"Filter generation failed: {e}")
return
print("\n=== GENERATED FILTER REQUEST ===")
print(json.dumps(filter_request, indent=2, ensure_ascii=False))
errors = validate_filter_request(filter_request)
if errors:
print("\nASK FAILED")
for err in errors:
print(f"- {err}")
return
mode = filter_request["mode"]
filter_obj = filter_request["filter"]
if mode == "all":
print("\n=== MODE ===")
print("all")
cmd_analyze()
return
print("\n=== MODE ===")
print("filter")
run_query_filter(filter_obj)
# --------------------------------------------------
# 3.9 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
cmd_ingest(sys.argv[2])
elif command == "analyze":
cmd_analyze()
elif command == "query":
if len(sys.argv) < 3:
print("Missing JSON filter for query.\n")
print_usage()
return
cmd_query(sys.argv[2])
elif command == "ask":
if len(sys.argv) < 3:
print("Missing natural language query for ask.\n")
print_usage()
return
cmd_ask(sys.argv[2])
else:
print(f"Unknown command: {command}\n")
print_usage()
if __name__ == "__main__":
main()