随机文本的URL缩短

代码

test.py

import random
import string
from urllib.parse import urlparse

class UrlShortener:

    def __init__(self):
        self.data_helper = DataHelperJson()
        self.conflict_num = 0

    @staticmethod
    def _get_random_text(size: int = 3):
        characters = string.ascii_letters + string.digits
        return ''.join(random.choice(characters) for _ in range(size))

    @staticmethod
    def _pack_url_item(url: str, random_text: str) -> dict:
        return {'url': url, 'random_text': random_text}

    async def _add_new_url(self, url: str) -> dict:
        random_text = self._get_random_text()

        while True:
            is_exists, _ = await self.data_helper.is_hash_exists(random_text)
            if is_exists:
                self.conflict_num += 1
                random_text = self._get_random_text()
            else:
                break

        url_item = self._pack_url_item(url, random_text)
        await self.data_helper.add_url_item(url_item)
        return url_item

    @staticmethod
    def _get_url_path(url):
        parsed_url = urlparse(url)
        path = parsed_url.path
        query = parsed_url.query
        return f"{path}?{query}" if query else path

    async def get(self, url: str):
        url = self._get_url_path(url)
        is_exists, url_item = await self.data_helper.is_url_exists(url)
        if not is_exists:
            url_item = await self._add_new_url(url)
        return url_item

    async def parse(self, random_text: str) -> (bool, dict):
        return await self.data_helper.is_hash_exists(random_text)


def main():
    usr = UrlShortener()

    print('URL数据量:', usr.data_helper.get_size())

    for i, item in enumerate(range(1000)):
        url = "Hello, World!" + str(item)
        url_item = usr.get(url)
        # print(url_item)

    print('生成映射数:', i + 1)
    print('URL冲突次数:', usr.conflict_num)

    print('URL数据量:', usr.data_helper.get_size())

if __name__ == "__main__":
    main()

data_helper.py

import os
import json

class DataHelperJson:

    def __init__(self, data_name='data_source.json'):
        self.data_name = data_name
        self.url_map = self._load_url_map()

    def _load_url_map(self, default_result=None) -> list:
        default_result = default_result or []

        def read_json_file(filename, default_result=None):
            default_result = default_result or []
            if os.path.exists(filename):
                try:
                    with open(filename, 'r', encoding='utf-8') as fp:
                        return json.load(fp)
                except json.JSONDecodeError:
                    return default_result
            else:
                return default_result

        return read_json_file(self.data_name, default_result)

    def _dump_url_map(self):
        def write_json_file(filename, data):
            with open(filename, 'w', encoding='utf-8') as fp:
                json.dump(data, fp)

        write_json_file(self.data_name, self.url_map)

    def is_url_exists(self, url: str) -> (bool, dict):
        for url_item in self.url_map:
            if url_item.get('url') == url:
                return True, url_item
        return False, None

    def is_hash_exists(self, random_text: str) -> (bool, dict):
        for url_item in self.url_map:
            if url_item.get('random_text') == random_text:
                return True, url_item
        return False, None

    def add_url_item(self, url_item: dict):
        self.url_map.append(url_item)
        print(f'[+] Added new item: {url_item}')
        self._dump_url_map()

    def get_size(self) -> int:
        return len(self.url_map)