gpt4free/g4f/cli/client.py

#!/usr/bin/env python3

import os
import sys
import asyncio
import json
import argparse
import traceback
import requests

from pathlib import Path
from typing import Optional, List, Dict
from g4f.client import AsyncClient
from g4f.providers.response import JsonConversation, MediaResponse, is_content
from g4f.cookies import set_cookies_dir, read_cookie_files
from g4f.Provider import ProviderUtils
from g4f.image import extract_data_uri, is_accepted_format
from g4f.image.copy_images import get_media_dir
from g4f.client.helper import filter_markdown
from g4f.errors import MissingRequirementsError

try:
    from g4f.integration.markitdown import MarkItDown
    has_markitdown = True
except ImportError:
    has_markitdown = False

from g4f.config import CONFIG_DIR, COOKIES_DIR
from g4f import debug

CONVERSATION_FILE = CONFIG_DIR / "conversation.json"


class ConversationManager:
    """Manages conversation history and state."""
    def __init__(
        self,
        file_path: Optional[Path] = None,
        model: Optional[str] = None,
        provider: Optional[str] = None,
        max_messages: int = 5
    ) -> None:
        self.file_path = file_path
        self.model = model
        self.provider = provider
        self.max_messages = max_messages
        self.conversation: Optional[JsonConversation] = None
        self.history: List[Dict[str, str]] = []
        self.data: Dict = {}
        self._load()

    def _load(self) -> None:
        if not self.file_path or not self.file_path.is_file():
            return
        try:
            with open(self.file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            if self.model is None:
                self.model = data.get("model")
            if self.provider is None:
                self.provider = data.get("provider")
            self.data = data.get("data", {})
            if self.provider and self.data.get(self.provider):
                self.conversation = JsonConversation(**self.data[self.provider])
            elif not self.provider and self.data:
                self.conversation = JsonConversation(**self.data)
            self.history = data.get("items", [])
        except Exception as e:
            print(f"Error loading conversation: {e}", file=sys.stderr)

    def save(self) -> None:
        if not self.file_path:
            return
        try:
            if self.conversation and self.provider:
                self.data[self.provider] = self.conversation.get_dict()
            elif self.conversation:
                self.data.update(self.conversation.get_dict())
            payload = {
                "model": self.model,
                "provider": self.provider,
                "data": self.data,
                "items": self.history
            }
            with open(self.file_path, 'w', encoding='utf-8') as f:
                json.dump(payload, f, indent=2, ensure_ascii=False)
        except Exception as e:
            print(f"Error saving conversation: {e}", file=sys.stderr)

    def add_message(self, role: str, content: str) -> None:
        self.history.append({"role": role, "content": content})

    def get_messages(self) -> List[Dict[str, str]]:
        result = []
        for item in self.history[-self.max_messages:]:
            if item.get("role") in ["user", "system"] or result:
                result.append(item)
        return result

async def stream_response(
    client: AsyncClient,
    input_text,
    conversation: ConversationManager,
    output_file: Optional[Path] = None,
    instructions: Optional[str] = None
) -> None:
    media = None
    if isinstance(input_text, tuple):
        media, input_text = input_text

    if instructions:
        conversation.add_message("system", instructions)

    conversation.add_message("user", input_text)

    create_args = {
        "model": conversation.model,
        "messages": conversation.get_messages(),
        "stream": True,
        "media": media,
        "conversation": conversation.conversation,
    }

    response_tokens = []
    last_chunk = None
    async for chunk in client.chat.completions.create(**create_args):
        last_chunk = chunk
        delta = chunk.choices[0].delta.content
        if not delta:
            continue
        if is_content(delta):
            response_tokens.append(delta)
        try:
            print(delta, end="", flush=True)
        except UnicodeEncodeError as e:
            debug.error(e)
            pass
    print()

    if last_chunk and hasattr(last_chunk, "conversation"):
        conversation.conversation = last_chunk.conversation

    media_chunk = next((t for t in response_tokens if isinstance(t, MediaResponse)), None)
    text_response = ""
    if media_chunk:
        text_response = response_tokens[0] if len(response_tokens) == 1 else "".join(str(t) for t in response_tokens)
    else:
        text_response = "".join(str(t) for t in response_tokens)

    if output_file:
        if save_content(text_response, media_chunk, str(output_file)):
            print(f"\n→ Response saved to '{output_file}'")

    if text_response:
        conversation.add_message("assistant", text_response)
    else:
        raise RuntimeError("No response received")


def save_content(content, media: Optional[MediaResponse], filepath: str, allowed_types=None) -> bool:
    if media:
        for url in media.urls:
            if url.startswith(("http://", "https://")):
                try:
                    resp = requests.get(url, cookies=media.get("cookies"), headers=media.get("headers"))
                    if resp.status_code == 200:
                        with open(filepath, "wb") as f:
                            f.write(resp.content)
                        return True
                except Exception as e:
                    print(f"Error fetching media '{url}': {e}", file=sys.stderr)
                return False
            else:
                content = url
                break
    if hasattr(content, "data"):
        content = content.data
    if not content:
        print("\nNo content to save.", file=sys.stderr)
        return False
    if content.startswith("data:"):
        with open(filepath, "wb") as f:
            f.write(extract_data_uri(content))
        return True
    if content.startswith("/media/"):
        src = content.replace("/media", get_media_dir()).split("?")[0]
        os.rename(src, filepath)
        return True
    filtered = filter_markdown(content, allowed_types)
    if filtered:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(filtered)
        return True
    print("\nUnable to save content.", file=sys.stderr)
    return False

def get_parser(exit_on_error=True):
    parser = argparse.ArgumentParser(
        description="G4F CLI client with conversation history",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        exit_on_error=exit_on_error
    )
    parser.add_argument('-d', '--debug', action='store_true', help="Verbose debug")
    parser.add_argument('-p', '--provider', default=None,
        help=f"Provider to use: {', '.join(k for k,v in ProviderUtils.convert.items() if v.working)}")
    parser.add_argument('-m', '--model', help="Model name")
    parser.add_argument('-O', '--output', type=Path,
        help="Save assistant output to FILE (text or media)")
    parser.add_argument('-i', '--instructions', help="System instructions")
    parser.add_argument('-c', '--cookies-dir', type=Path, default=COOKIES_DIR,
        help="Cookies/HAR directory")
    parser.add_argument('--conversation-file', type=Path, default=CONVERSATION_FILE,
        help="Conversation JSON")
    parser.add_argument('-C', '--clear-history', action='store_true', help="Wipe history")
    parser.add_argument('-N', '--no-config', action='store_true', help="Skip loading history")
    # <-- updated -e/--edit to take an optional filename
    parser.add_argument(
        '-e', '--edit',
        type=Path,
        metavar='FILE',
        help="If FILE given: send its contents and overwrite it with AI's reply."
    )
    parser.add_argument('--max-messages', type=int, default=5,
        help="Max user+assistant turns in context")
    parser.add_argument('input', nargs='*',
        help="URLs, image paths or plain text")
    return parser


async def run_args(input_val, args):
    try:
        # ensure dirs
        if args.output:
            args.output.parent.mkdir(parents=True, exist_ok=True)
        if args.conversation_file:
            args.conversation_file.parent.mkdir(parents=True, exist_ok=True)
        args.cookies_dir.mkdir(parents=True, exist_ok=True)

        if args.debug:
            debug.logging = True

        conv = ConversationManager(
            None if args.no_config else args.conversation_file,
            model=args.model,
            provider=args.provider,
            max_messages=args.max_messages
        )
        if args.clear_history:
            conv.history = []
            conv.conversation = None

        set_cookies_dir(str(args.cookies_dir))
        read_cookie_files()

        client = AsyncClient(provider=conv.provider)

        if isinstance(args.edit, Path):
            file_to_edit = args.edit
            if not file_to_edit.exists():
                print(f"ERROR: file not found: {file_to_edit}", file=sys.stderr)
                sys.exit(1)
            text = file_to_edit.read_text(encoding="utf-8")
            # we will both send and overwrite this file
            input_val = f"```file: {file_to_edit}\n{text}\n```\n" + (input_val[1] if isinstance(input_val, tuple) else input_val)
            output_target = file_to_edit
        else:
            # normal, non-edit mode
            output_target = args.output

        await stream_response(client, input_val, conv, output_target, args.instructions)
        conv.save()

    except Exception:
        print(traceback.format_exc(), file=sys.stderr)
        sys.exit(1)


def run_client_args(args, exit_on_error=True):
    input_txt = ""
    media = []
    rest = 0

    for idx, tok in enumerate(args.input):
        if tok.startswith(("http://","https://")):
            # same URL logic...
            resp = requests.head(tok, allow_redirects=True)
            if resp.ok and resp.headers.get("Content-Type","").startswith("image"):
                media.append(tok)
            else:
                if not has_markitdown:
                    raise MissingRequirementsError("Install markitdown")
                md = MarkItDown()
                txt = md.convert_url(tok).text_content
                input_txt += f"\n```source: {tok}\n{txt}\n```\n"
        elif os.path.isfile(tok):
            head = Path(tok).read_bytes()[:12]
            try:
                if is_accepted_format(head):
                    media.append(Path(tok))
                    is_img = True
                else:
                    is_img = False
            except ValueError:
                is_img = False
            if not is_img:
                txt = Path(tok).read_text(encoding="utf-8")
                input_txt += f"\n```file: {tok}\n{txt}\n```\n"
        else:
            rest = idx
            break
        rest = idx + 1

    tail = args.input[rest:]
    if tail:
        input_txt = " ".join(tail) + "\n" + input_txt

    if not sys.stdin.isatty() and not input_txt:
        input_txt = sys.stdin.read()

    if media:
        val = (media, input_txt)
    else:
        val = input_txt.strip()

    if exit_on_error and not val:
        print("No input provided. Use -h.", file=sys.stderr)
        sys.exit(1)
    elif not val:
        raise argparse.ArgumentError(None, "No input provided. Use -h for help.")

    asyncio.run(run_args(val, args))


if __name__ == "__main__":
    run_client_args(get_parser().parse_args())