113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
from services.hubstaff_api import HubstaffAPI
|
|
import os
|
|
import shutil
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
async def main():
|
|
# Load the Hubstaff API key from environment variables
|
|
api_key = os.getenv("HUBSTAFF_API_KEY")
|
|
if not api_key:
|
|
print("Error: HUBSTAFF_API_KEY environment variable not set.")
|
|
return
|
|
|
|
# Load the Hubstaff organization ID from environment variables
|
|
org_id = os.getenv("HUBSTAFF_ORG_ID")
|
|
if not org_id:
|
|
print("Error: HUBSTAFF_ORG_ID environment variable not set.")
|
|
return
|
|
|
|
# Initialize the Hubstaff API client
|
|
hubstaff = HubstaffAPI(refresh_token=api_key)
|
|
|
|
try:
|
|
# Fetch user info
|
|
user_info = await hubstaff.get_current_user()
|
|
user_id = user_info["user"]["id"]
|
|
|
|
# Define the date range for the timesheet
|
|
end_date = datetime.now()
|
|
start_date = end_date - timedelta(days=7)
|
|
|
|
# Fetch timesheet data
|
|
timesheet_data = await hubstaff.get_timesheets(
|
|
user_id, start=start_date.isoformat(), stop=end_date.isoformat()
|
|
)
|
|
|
|
# Check if timesheet_data is valid and "times" is a list before proceeding
|
|
times = timesheet_data.get("times") if timesheet_data else None
|
|
if not times or not isinstance(times, list):
|
|
print("Error: No timesheet data found or invalid format received.")
|
|
return
|
|
|
|
# Prepare invoice configuration based on timesheet data
|
|
invoice_logo = os.getenv("INVOICE_LOGO", "/path/to/logo.png")
|
|
invoice_from = os.getenv("INVOICE_FROM", "Default Company")
|
|
invoice_to = os.getenv("INVOICE_TO", "Client Company")
|
|
invoice_tax = float(os.getenv("INVOICE_TAX", "0.0"))
|
|
invoice_rate = float(os.getenv("INVOICE_RATE", "25"))
|
|
|
|
config = {
|
|
"logo": invoice_logo,
|
|
"from": invoice_from,
|
|
"to": invoice_to,
|
|
"tax": invoice_tax,
|
|
"items": [],
|
|
"quantities": [],
|
|
"rates": [],
|
|
}
|
|
|
|
# For each timesheet entry, we'll use the note as item.
|
|
# If note is missing or "N/A", we'll use the date instead.
|
|
for entry in times:
|
|
entry_date = datetime.fromtimestamp(int(entry["date"])).strftime("%Y-%m-%d")
|
|
note = entry.get("note", "N/A")
|
|
item = note if note != "N/A" else f"Work on {entry_date}"
|
|
minutes = entry.get("minutes", 0)
|
|
config["items"].append(item)
|
|
config["quantities"].append(minutes)
|
|
config["rates"].append(invoice_rate)
|
|
|
|
# Generate YAML output
|
|
yaml_lines = [
|
|
f"logo: {config['logo']}",
|
|
f"from: {config['from']}",
|
|
f"to: {config['to']}",
|
|
f"tax: {config['tax']}",
|
|
"items:",
|
|
]
|
|
for item in config["items"]:
|
|
yaml_lines.append(f' - "{item}"')
|
|
yaml_lines.append("quantities:")
|
|
for qty in config["quantities"]:
|
|
yaml_lines.append(f" - {qty}")
|
|
yaml_lines.append("rates:")
|
|
for rate in config["rates"]:
|
|
yaml_lines.append(f" - {rate}")
|
|
|
|
yaml_content = "\n".join(yaml_lines)
|
|
config_filename = "invoice_config.yaml"
|
|
|
|
with open(config_filename, "w") as f:
|
|
f.write(yaml_content)
|
|
|
|
print(f"Invoice configuration written to {config_filename}")
|
|
|
|
# Call the invoice command if available locally
|
|
if shutil.which("invoice"):
|
|
os.system(
|
|
f"invoice generate --import {config_filename} --output timesheet-invoice.pdf"
|
|
)
|
|
else:
|
|
print(
|
|
"Invoice command not found locally. Please install it to generate invoices."
|
|
)
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching timesheet data: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|