182 lines
6.1 KiB
Python
182 lines
6.1 KiB
Python
#!/usr/bin/env python3
|
|
"""CLI for managing food log entries in MongoDB lists.food_log."""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone, timedelta
|
|
from pymongo import MongoClient
|
|
|
|
CONN = "mongodb://root:3wwfoUjyk2E2zWELXFlLuHqfw1ALlOp4pb2H5Vq3TImbMIHL2h1u8Jej2mjzCPl@docdb.connorrhodes.com:35563/?tls=true&tlsAllowInvalidCertificates=true"
|
|
|
|
def get_collection():
|
|
client = MongoClient(CONN)
|
|
return client["lists"]["food_log"], client
|
|
|
|
def cmd_add(args):
|
|
coll, client = get_collection()
|
|
doc = {"timestamp": datetime.now(timezone.utc)}
|
|
|
|
if args.name:
|
|
doc["name"] = args.name
|
|
if args.type:
|
|
doc["type"] = args.type
|
|
if args.amount:
|
|
doc["amount"] = args.amount
|
|
if args.unit:
|
|
doc["unit"] = args.unit
|
|
if args.extra:
|
|
try:
|
|
extra = json.loads(args.extra)
|
|
doc.update(extra)
|
|
except json.JSONDecodeError:
|
|
print(f"Error: --extra must be valid JSON. Got: {args.extra}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if "name" not in doc:
|
|
print("Error: --name is required", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
result = coll.insert_one(doc)
|
|
out = coll.find_one({"_id": result.inserted_id})
|
|
out["_id"] = str(out["_id"])
|
|
print(json.dumps(out, indent=2, default=str))
|
|
client.close()
|
|
|
|
def cmd_get(args):
|
|
coll, client = get_collection()
|
|
query = {}
|
|
|
|
if args.name:
|
|
query["name"] = {"$regex": args.name, "$options": "i"}
|
|
if args.type:
|
|
query["type"] = {"$regex": args.type, "$options": "i"}
|
|
if args.today:
|
|
local_now = datetime.now().astimezone()
|
|
local_midnight = local_now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
today_start = local_midnight.astimezone(timezone.utc)
|
|
query["timestamp"] = {"$gte": today_start}
|
|
if args.days:
|
|
local_now = datetime.now().astimezone()
|
|
cutoff_local = local_now - timedelta(days=args.days)
|
|
cutoff = cutoff_local.astimezone(timezone.utc)
|
|
query["timestamp"] = {"$gte": cutoff}
|
|
|
|
cursor = coll.find(query).sort("timestamp", -1).limit(args.limit if args.limit else 50)
|
|
docs = []
|
|
for d in cursor:
|
|
d["_id"] = str(d["_id"])
|
|
docs.append(d)
|
|
|
|
if not docs:
|
|
print("No entries found.")
|
|
else:
|
|
print(json.dumps(docs, indent=2, default=str))
|
|
client.close()
|
|
|
|
def cmd_update(args):
|
|
coll, client = get_collection()
|
|
|
|
from bson import ObjectId
|
|
|
|
if not args.id:
|
|
print("Error: --id is required for update", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
query = {"_id": ObjectId(args.id)}
|
|
|
|
update_fields = {}
|
|
if args.set_name:
|
|
update_fields["name"] = args.set_name
|
|
if args.set_type:
|
|
update_fields["type"] = args.set_type
|
|
if args.set_amount is not None:
|
|
update_fields["amount"] = args.set_amount
|
|
if args.set_unit:
|
|
update_fields["unit"] = args.set_unit
|
|
if args.set_extra:
|
|
try:
|
|
extra = json.loads(args.set_extra)
|
|
update_fields.update(extra)
|
|
except json.JSONDecodeError:
|
|
print(f"Error: --set-extra must be valid JSON", file=sys.stderr)
|
|
sys.exit(1)
|
|
if args.unset:
|
|
for field in args.unset:
|
|
update_fields[field] = ""
|
|
|
|
if not update_fields:
|
|
print("Error: nothing to update", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
set_doc = {"$set": {k: v for k, v in update_fields.items()}}
|
|
unset_doc = {"$unset": {k: v for k, v in update_fields.items() if v == ""}}
|
|
|
|
update = {}
|
|
if set_doc["$set"]:
|
|
update["$set"] = set_doc["$set"]
|
|
if unset_doc["$unset"]:
|
|
update["$unset"] = unset_doc["$unset"]
|
|
|
|
result = coll.update_one(query, update)
|
|
print(f"Matched: {result.matched_count}, Modified: {result.modified_count}")
|
|
|
|
if result.matched_count:
|
|
updated = coll.find_one(query)
|
|
updated["_id"] = str(updated["_id"])
|
|
print(json.dumps(updated, indent=2, default=str))
|
|
else:
|
|
print("No matching entry found.")
|
|
client.close()
|
|
|
|
def cmd_delete(args):
|
|
from bson import ObjectId
|
|
coll, client = get_collection()
|
|
|
|
if not args.id:
|
|
print("Error: --id is required for delete", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
result = coll.delete_one({"_id": ObjectId(args.id)})
|
|
print(f"Deleted: {result.deleted_count}")
|
|
client.close()
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Manage food log entries in lists.food_log")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# add
|
|
p_add = sub.add_parser("add", help="Add a food log entry")
|
|
p_add.add_argument("--name", required=True)
|
|
p_add.add_argument("--type")
|
|
p_add.add_argument("--amount", type=float)
|
|
p_add.add_argument("--unit")
|
|
p_add.add_argument("--extra", help="JSON string with additional fields")
|
|
|
|
# get
|
|
p_get = sub.add_parser("get", help="Get food log entries")
|
|
p_get.add_argument("--name", help="Filter by name (regex)")
|
|
p_get.add_argument("--type", help="Filter by type (regex)")
|
|
p_get.add_argument("--today", action="store_true", help="Only today's entries")
|
|
p_get.add_argument("--days", type=int, help="Entries from last N days")
|
|
p_get.add_argument("--limit", type=int, help="Max entries to return")
|
|
|
|
# update
|
|
p_upd = sub.add_parser("update", help="Update a food log entry")
|
|
p_upd.add_argument("--id", required=True, help="ObjectId to update")
|
|
p_upd.add_argument("--set-name", dest="set_name")
|
|
p_upd.add_argument("--set-type", dest="set_type")
|
|
p_upd.add_argument("--set-amount", dest="set_amount", type=float)
|
|
p_upd.add_argument("--set-unit", dest="set_unit")
|
|
p_upd.add_argument("--set-extra", dest="set_extra", help="JSON string with additional fields to set")
|
|
p_upd.add_argument("--unset", nargs="*", help="Field names to remove")
|
|
|
|
# delete
|
|
p_del = sub.add_parser("delete", help="Delete a food log entry")
|
|
p_del.add_argument("--id", required=True, help="ObjectId to delete")
|
|
|
|
args = parser.parse_args()
|
|
{"add": cmd_add, "get": cmd_get, "update": cmd_update, "delete": cmd_delete}[args.command](args)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|