llm_ticket3/utils/ticket_manager2.py

import os
import json
import base64
import requests
from typing import Dict, List, Any, Optional
from datetime import datetime

class TicketManager:
    def __init__(self, url: str, db: str, username: str, api_key: str):
        self.url = url
        self.db = db
        self.username = username
        self.api_key = api_key
        self.uid = None
        self.session_id = None
        self.model_name = "project.task"
        
    def login(self) -> bool:
        login_url = f"{self.url}/web/session/authenticate"
        login_data = {
            "jsonrpc": "2.0",
            "params": {
                "db": self.db,
                "login": self.username,
                "password": self.api_key
            }
        }
        response = requests.post(login_url, json=login_data)
        result = response.json()
        
        if result.get("error"):
            print(f"Erreur de connexion: {result['error']['message']}")
            return False
        
        self.uid = result.get("result", {}).get("uid")
        self.session_id = response.cookies.get("session_id")
        return True if self.uid else False
    
    def _rpc_call(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
        full_url = f"{self.url}{endpoint}"
        headers = {"Content-Type": "application/json"}
        data = {"jsonrpc": "2.0", "method": "call", "params": params}
        
        response = requests.post(full_url, json=data, headers=headers, cookies={"session_id": self.session_id})
        return response.json().get("result", {})
    
    def get_ticket(self, ticket_id: int) -> Dict[str, Any]:
        params = {
            "model": self.model_name,
            "method": "read",
            "args": [[ticket_id]],
            "kwargs": {}
        }
        return self._rpc_call("/web/dataset/call_kw", params)
           
    def get_ticket_messages(self, ticket_id: int) -> List[Dict[str, Any]]:
        messages = self._rpc_call("/web/dataset/call_kw", {
            "model": "mail.message",
            "method": "search_read",
            "args": [[["res_id", "=", ticket_id], ["model", "=", "project.task"]]],
            "kwargs": {"fields": ["id", "body", "author_id", "date"]}
        })
        return messages
    
    def get_ticket_by_code(self, ticket_code: str) -> Dict[str, Any]:
        """
        Récupère un ticket par son code unique.
    
     Args:
         ticket_code: Code du ticket à récupérer (par exemple, T11067)
    
     Returns:
            Dictionnaire contenant les informations du ticket ou une erreur si non trouvé
        """
        # Recherche du ticket par code
        params = {
            "model": self.model_name,
            "method": "search_read",
            "args": [[["code", "=", ticket_code]], ["id", "name", "description", "stage_id"]],
            "kwargs": {"limit": 1}
        }
    
        result = self._rpc_call("/web/dataset/call_kw", params)
    
        if not result:
            print(f"Aucun ticket trouvé avec le code {ticket_code}")
            return {}
    
        # Retourne le premier ticket trouvé
        return result[0] if isinstance(result, list) and len(result) > 0 else {}
    
    def save_json(self, data: Any, path: str):
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
    
    def extract_ticket_data(self, ticket_id: int, output_dir: str):
        os.makedirs(output_dir, exist_ok=True)
        
        ticket_data = self.get_ticket(ticket_id)
        self.save_json(ticket_data, os.path.join(output_dir, "ticket_info.json"))
        
        messages = self.get_ticket_messages(ticket_id)
        
        # Save messages in raw and cleaned formats
        self.save_json(messages, os.path.join(output_dir, "messages_raw.json"))
        
        cleaned_messages = []
        for msg in messages:
            cleaned_messages.append({
                "message_id": msg["id"],
                "sender": msg["author_id"][1] if msg["author_id"] else "Unknown",
                "timestamp": msg["date"],
                "content": self.clean_html(msg["body"])
            })
        
        self.save_json(cleaned_messages, os.path.join(output_dir, "all_messages.json"))
        
        # Generate structure.json
        structure = {
            "date_extraction": datetime.now().isoformat(),
            "ticket_dir": output_dir,
            "fichiers_json": [
                "ticket_info.json",
                "messages_raw.json",
                "all_messages.json"
            ]
        }
        self.save_json(structure, os.path.join(output_dir, "structure.json"))
    
    def clean_html(self, html_content: str) -> str:
        import re
        from html import unescape
        
        text = re.sub(r'<.*?>', '', html_content)
        text = unescape(text)
        text = re.sub(r'\s+', ' ', text).strip()
        
        return text

if __name__ == "__main__":
    import sys
    
    if len(sys.argv) < 2:
        print("Usage: python ticket_manager2.py <ticket_id>")
        sys.exit(1)
    
    ticket_id = int(sys.argv[1])
    
    # Configuration Odoo
    url = "https://odoo.cbao.fr"
    db = "database_name"
    username = "username"
    api_key = "api_key"
    
    manager = TicketManager(url, db, username, api_key)
    
    if not manager.login():
        print("Échec de connexion à Odoo")
        sys.exit(1)
    
    output_dir = f"T11067_analysis/ticket_structure"
    manager.extract_ticket_data(ticket_id, output_dir)
    
    print("Extraction terminée avec succès.")