Add pixiv_json_to_html

.gitignore

@@ -141,3 +141,5 @@ dmypy.json
 *.yml
 *.conf
 !*.example.yaml
+*.jsonl
+tiktokencache

chat.py

@@ -0,0 +1,212 @@
+import openai
+import httpx
+import yaml
+import argparse
+import asyncio
+import json
+from typing import Optional
+
+
+class Config:
+    def __init__(self, args, yaml_config):
+        self._args = args
+        self._cfg = cfg
+
+    @property
+    def model(self) -> str:
+        if self._args.model:
+            return self._args.model
+        if 'model' in self._cfg and isinstance(self._cfg['model'], str):
+            return self._cfg['model']
+        return 'gpt-4o-mini'
+
+    @property
+    def max_completion_tokens(self) -> int:
+        if self._args.max_completion_tokens:
+            return self._args.max_completion_tokens
+        if 'max_completion_tokens' in self._cfg and isinstance(self._cfg['max_completion_tokens'], int):  # noqa: E501
+            return self._cfg['max_completion_tokens']
+        return 4096
+
+    @property
+    def include_usage(self) -> bool:
+        if self._args.include_usage:
+            return self._args.include_usage
+        if 'include_usage' in self._cfg and isinstance(self._cfg['include_usage'], bool):  # noqa: E501
+            return self._cfg['include_usage']
+        return False
+
+    @property
+    def output(self) -> Optional[str]:
+        if self._args.output:
+            return self._args.output
+        if 'output' in self._cfg and isinstance(self._cfg['output'], str):
+            return self._cfg['output']
+        return None
+
+    @property
+    def temperature(self) -> float:
+        if self._args.temperature:
+            if self._args.temperature >= 0.0 and self._args.temperature <= 2.0:
+                return self._args.temperature
+        if 'temperature' in self._cfg and isinstance(self._cfg['temperature'], float):  # noqa: E501
+            temperature = self._cfg['temperature']
+            if temperature >= 0.0 and temperature <= 2.0:
+                return temperature
+        return 1.0
+
+    @property
+    def top_p(self) -> float:
+        if self._args.top_p:
+            if self._args.top_p >= 0.0 and self._args.top_p <= 1.0:
+                return self._args.top_p
+        if 'top_p' in self._cfg and isinstance(self._cfg['top_p'], float):
+            top_p = self._cfg['top_p']
+            if top_p >= 0.0 and top_p <= 1.0:
+                return top_p
+        return 1.0
+
+    @property
+    def presence_penalty(self) -> float:
+        if self._args.presence_penalty:
+            if self._args.presence_penalty >= -2.0 and self._args.presence_penalty <= 2.0:  # noqa: E501
+                return self._args.presence_penalty
+        if 'presence_penalty' in self._cfg and isinstance(self._cfg['presence_penalty'], float):  # noqa: E501
+            presence_penalty = self._cfg['presence_penalty']
+            if presence_penalty >= -2.0 and presence_penalty <= 2.0:
+                return presence_penalty
+        return 0.0
+
+    @property
+    def store(self) -> bool:
+        if self._args.store:
+            return self._args.store
+        if 'store' in self._cfg and isinstance(self._cfg['store'], bool):
+            return self._cfg['store']
+        return False
+
+    @property
+    def proxy(self) -> Optional[str]:
+        if self._args.proxy:
+            return self._args.proxy
+        if 'proxy' in self._cfg and isinstance(self._cfg['proxy'], str):
+            return self._cfg['proxy']
+        return None
+
+    @property
+    def skip_verify(self) -> bool:
+        if self._args.skip_verify:
+            return self._args.skip_verify
+        if 'skip_verify' in self._cfg and isinstance(self._cfg['skip_verify'], bool):  # noqa: E501
+            return self._cfg['skip_verify']
+        return False
+
+
+def load_config(file_path):
+    with open(file_path, 'r', encoding='utf-8') as file:
+        config = yaml.safe_load(file)
+    openai.api_key = config['api_key']
+    if 'base_url' in config:
+        openai.base_url = config['base_url']
+    return config
+
+
+def get_user_prompt():
+    prompt = ""
+    print("Enter your prompt (leave empty to finish):")
+    while True:
+        line = input('')
+        if line.strip() == "":
+            break
+        prompt += line + "\n"
+    return prompt.strip()
+
+
+async def stream_response(messages, prompt, args: Config):
+    cli = httpx.AsyncClient(proxy=args.proxy, verify=not args.skip_verify)
+    client = openai.AsyncClient(api_key=openai.api_key,
+                                base_url=openai.base_url,
+                                http_client=cli)
+    messages.append({"role": "user", "content": prompt})
+    response = await client.chat.completions.create(
+        model=args.model,
+        max_completion_tokens=args.max_completion_tokens,
+        messages=messages,
+        stream_options={"include_usage": args.include_usage},
+        temperature=args.temperature,
+        top_p=args.top_p,
+        presence_penalty=args.presence_penalty,
+        store=args.store,
+        stream=True
+    )
+    res = ''
+    thinking = -1
+    async for chunk in response:
+        if chunk.choices:
+            choice = chunk.choices[0]
+            if choice.delta:
+                if choice.delta.reasoning_content:
+                    if thinking == -1:
+                        thinking = 0
+                        print('Start thinking')
+                    print(choice.delta.reasoning_content, end='', flush=True)
+                if choice.delta.content:
+                    if thinking == 0:
+                        thinking = 1
+                        print('End thinking')
+                    data = choice.delta.content
+                    res += data
+                    print(data, end='', flush=True)
+    print(flush=True)
+    if chunk.usage:
+        print(f"Usage: {chunk.usage.to_json(indent=None)}")
+    return {'role': 'assistant', 'content': res}
+
+
+async def chat(args: Config):
+    messages = []
+    while True:
+        try:
+            user_prompt = get_user_prompt()
+        except KeyboardInterrupt:
+            break
+        while True:
+            try:
+                res = await stream_response(messages, user_prompt, args)
+                messages.append(res)
+                break
+            except KeyboardInterrupt:
+                break
+            except openai.InternalServerError:
+                print("Internal server error, retrying...")
+                continue
+    if args.output and len(messages):
+        save_to_jsonl(args.output, messages)
+
+
+def save_to_jsonl(file_path, message):
+    with open(file_path, 'a', encoding='utf-8') as f:
+        json.dump({'messages': message}, f, ensure_ascii=False,
+                  separators=(',', ':'))
+        f.write('\n')
+
+
+parser = argparse.ArgumentParser(description="Chat with OpenAI's model.")
+parser.add_argument('-m', '--model', type=str, help='Chat model to use')  # noqa: E501
+parser.add_argument('-M', '--max-completion-tokens', type=int, help='Maximum length of the response')  # noqa: E501
+parser.add_argument('-c', '--config', type=str, default='./chat.yml', help='Path to the configuration file')  # noqa: E501
+parser.add_argument('-o', '--output', type=str, help='Path to the output JSONL file')  # noqa: E501
+parser.add_argument('-i', '--include-usage', action='store_true', help='Include usage information in the response')  # noqa: E501
+parser.add_argument('-t', '--temperature', type=float, help='What sampling temperature to use, between 0 and 2. Higher values like 0.8 will make the output more random, while lower values like 0.2 will make it more focused and deterministic. ')  # noqa: E501
+parser.add_argument('-p', '--top-p', type=float, help='An alternative to sampling with temperature, called nucleus sampling, where the model considers the results of the tokens with top_p probability mass. So 0.1 means only the tokens comprising the top 10%% probability mass are considered.')  # noqa: E501
+parser.add_argument('-P', '--presence-penalty', type=float, help="Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far, increasing the model's likelihood to talk about new topics.")  # noqa: E501
+parser.add_argument('-s', '--store', action='store_true', help='Whether or not to store the output of this chat completion request for use in our model distillation or evals products.')  # noqa: E501
+parser.add_argument('-x', '--proxy', type=str, help='Proxy server URL to use for requests')  # noqa: E501
+parser.add_argument('-k', '--skip-verify', action='store_true', help='Skip SSL certificate verification for HTTPS requests')  # noqa: E501
+
+
+if __name__ == "__main__":
+    args = parser.parse_args()
+    cfg = load_config(args.config)
+    acfg = Config(args, cfg)
+    asyncio.run(chat(acfg))

convert_lrc.py

@@ -76,7 +76,7 @@ class Lyric:
                 re.append({'time': max(ltmp, round(i['time'] - 0.01, 2)),
                            'data': tmp})
                 tmp = None
-            elif i['time'] == ltmp:
+            elif i['time'] == ltmp and i['time'] != 0:
                 if tmp is None:
                     tmp = i['data']
                 else:

count_tokens.py

@@ -0,0 +1,45 @@
+import json
+import tiktoken
+import argparse
+
+
+def count_tokens(text, encoding):
+    return len(encoding.encode(text))
+
+
+def calculate_tokens_in_file(file_path, encoding_name, model_name, overhead):
+    if encoding_name:
+        encoding = tiktoken.get_encoding(encoding_name)
+    elif model_name:
+        encoding = tiktoken.encoding_for_model(model_name)
+    else:
+        encoding = tiktoken.get_encoding('cl100k_base')
+    print('Encoding name:', encoding.name)
+    total_tokens = 0
+
+    with open(file_path, 'r', encoding='utf-8') as file:
+        for line in file:
+            data = json.loads(line)
+            messages = data.get('messages', [])
+            for message in messages:
+                content = message.get('content', '')
+                tokens = count_tokens(content, encoding)
+                total_tokens += tokens + overhead
+    print(f"Total tokens in file: {total_tokens}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Calculate the number of tokens in a JSONL file.")  # noqa: E501
+    parser.add_argument('file_path', type=str, help='Path to the JSONL file')
+    parser.add_argument('-e', '--encoding', type=str, help='Encoding to use for tokenization (default: cl100k_base)')  # noqa: E501
+    parser.add_argument('-m', '--model', type=str, help='Encoding model')
+    parser.add_argument('-o', '--overhead', type=int, default=3, help='Overhead token count for each message. Default: 3')  # noqa: E501
+
+    args = parser.parse_args()
+
+    calculate_tokens_in_file(args.file_path, args.encoding, args.model,
+                             args.overhead)
+
+
+if __name__ == "__main__":
+    main()

pixiv_json_to_html.py

@@ -0,0 +1,125 @@
+from os.path import join, isfile, basename, abspath
+from json import load
+from html import escape
+
+
+def pixiv_json_to_html(dir_path):
+    json_path = join(dir_path, 'data.json')
+    with open(json_path, 'r', encoding='utf-8') as f:
+        data = load(f)
+    raw = data['raw']
+    title = raw['title']
+    html_data = f'<!DOCTYPE html><html><head><meta charset="utf-8"><title>{escape(title)}</title></head><body>\n'
+    html_data += f'<h1>{escape(title)}</h1>\n'
+    typ = raw['type']
+    img_idx = 1
+    used = set()
+    pbase = basename(abspath(dir_path))
+    if typ == 'article':
+        blocks = raw['body']['blocks']
+        imageMap = raw['body']['imageMap']
+        fileMap = raw['body']['fileMap']
+        urlEmbedMap = raw['body']['urlEmbedMap']
+        for block in blocks:
+            block_type = block['type']
+            if block_type == 'p':
+                text = block['text']
+                styles = []
+                if 'styles' in block:
+                    styles = block['styles']
+                events = []
+                for style in styles:
+                    if style['type'] == 'bold':
+                        start = style['offset']
+                        end = style['offset'] + style['length']
+                        events.append((start, '<b>', False))
+                        events.append((end, '</b>', True))
+                    else:
+                        raise ValueError(f'Unsupported style type: {style["type"]}')
+                events.sort(key=lambda x: (x[0], not x[2]))
+                output = ""
+                last_idx = 0
+                for pos, tag, is_closing in events:
+                    output += escape(text[last_idx:pos])
+                    output += tag
+                    last_idx = pos
+                output += escape(text[last_idx:])
+                html_data += f'<p>{output}</p>\n'
+            elif block_type == 'header':
+                for key in block:
+                    if key not in ['type', 'text']:
+                        print(block)
+                        raise ValueError(f'Unsupported header block key: {key}')
+                text = block['text']
+                html_data += f'<h2>{escape(text)}</h2>\n'
+            elif block_type == 'image':
+                image_id = block['imageId']
+                image_url = imageMap[image_id]['originalUrl']
+                image_ext = imageMap[image_id]['extension']
+                image_path = join(dir_path, f'{pbase}_{img_idx}.{image_ext}')
+                if not isfile(image_path):
+                    print(image_path)
+                    image_path = join(dir_path, f'{image_id}.{image_ext}')
+                image_path = basename(image_path)
+                html_data += f'<p><img src="{escape(image_path)}" data-original-url="{escape(image_url)}"></p>\n'
+                if image_id not in used:
+                    used.add(image_id)
+                    img_idx += 1
+            elif block_type == 'url_embed':
+                urlEmbedId = block['urlEmbedId']
+                embedData = urlEmbedMap[urlEmbedId]
+                embedType = embedData['type']
+                if embedType == 'fanbox.post':
+                    postInfo = embedData['postInfo']
+                    postTitle = postInfo['title']
+                    postUrl = f'https://www.fanbox.cc/{postInfo['creatorId']}/posts/{postInfo['id']}'
+                    postCover = None
+                    if 'cover' in postInfo and postInfo['cover']['type'] == 'cover_image':
+                        postCover = postInfo['cover']['url']
+                    postExcerpt = postInfo['excerpt']
+                    html_data += f'<div class="url-embed"><h2><a href="{escape(postUrl)}">{escape(postTitle)}</a></h2>'
+                    if postCover:
+                        html_data += f'<p><img src="{escape(postCover)}"></p>'
+                    html_data += f'<p>{escape(postExcerpt)}</p></div>\n'
+                elif embedType == 'html.card':
+                    embed_html_data = embedData['html']
+                    html_data += f'<div class="url-embed">{embed_html_data}</div>\n'
+                elif embedType == 'html':
+                    embed_html_data = embedData['html']
+                    html_data += f'<div class="url-embed">{embed_html_data}</div>\n'
+                elif embedType == 'default':
+                    url = embedData['url']
+                    html_data += f'<div class="url-embed"><iframe src="{escape(url)}" /></div>\n'
+                else:
+                    raise ValueError(f'Unsupported embed type: {embedType}')
+            else:
+                print(block)
+                raise ValueError(f'Unsupported block type: {block_type}')
+    elif typ == 'image':
+        text = raw['body']['text']
+        images = raw['body']['images']
+        html_data += f'<p>{escape(text).replace('\n', '<br>')}</p>\n'
+        for image in images:
+            image_id = image['id']
+            image_url = image['originalUrl']
+            image_ext = image['extension']
+            image_path = join(dir_path, f'{pbase}_{img_idx}.{image_ext}')
+            if not isfile(image_path):
+                print(image_path)
+                image_path = join(dir_path, f'{image_id}.{image_ext}')
+            image_path = basename(image_path)
+            html_data += f'<p><img src="{escape(image_path)}" data-original-url="{escape(image_url)}"></p>\n'
+            img_idx += 1
+    else:
+        raise ValueError(f'Unsupported body type: {typ}')
+    html_data += '</body></html>\n'
+    with open(join(dir_path, 'output.html'), 'w', encoding='utf-8') as f:
+        f.write(html_data)
+
+
+if __name__ == '__main__':
+    import argparse
+    parser = argparse.ArgumentParser(description='Convert Pixiv JSON to HTML')
+    parser.add_argument('dir', help='Directory containing data.json')
+    args = parser.parse_args()
+    pixiv_json_to_html(args.dir)