# !/usr/bin/python3 import logging import json import os from aiohttp import web import aiohttp_cors from multiprocessing import Process, cpu_count from socket import SOL_SOCKET, SO_REUSEADDR, socket import tempfile from urllib.parse import urlparse import sys import re from pathlib import Path from fontTools.subset import main as subset from fontTools.ttLib import TTFont from libs.app_tool import AppTool LISTEN_PORT = 8901 class FontServer: def __init__(self): self.app_tool = AppTool('font_ss.log') self.ZK_CONF = self.app_tool.zk() self.hostname = os.uname()[1] self.logger = logging self.base_dir = Path(__file__).parent def get_cache_path_by_url(self, url): parts = urlparse(url) tmp_dir = tempfile.gettempdir() font_dir = os.path.join(tmp_dir, "__xiuxiu_fonts_cache") if not os.path.isdir(font_dir): os.makedirs(os.path.join(tmp_dir, "__xiuxiu_fonts_cache")) filename = os.path.basename(parts.path) cached_file = os.path.join(font_dir, filename) return cached_file def download_font(self, url): cached_file = self.get_cache_path_by_url(url) if not os.path.exists(cached_file): sys.stderr.write('Downloading: "{}" to {}\n'.format(url, cached_file)) hash_prefix = None self.app_tool.download_url_to_file(url, cached_file, hash_prefix, progress=False) return cached_file def temp_path(self, suffix = None): tmp_dir = tempfile.gettempdir() subset_dir = os.path.join(tmp_dir, "__xiuxiu_fonts_subset_output") if not os.path.isdir(subset_dir): os.makedirs(os.path.join(tmp_dir, "__xiuxiu_fonts_subset_output")) file_name = self.app_tool.uuid() if suffix is not None: file_name += suffix return os.path.join(subset_dir, file_name), file_name def do_font_minify(self, font_url, with_txt = False, oss_bucket = None, obj_prefix = None): font_path = self.download_font(font_url) tmp_path, f1 = self.temp_path('.tff') tmp_path2, f1 = self.temp_path('.woff') if oss_bucket not in ['cs-rt', 'xiuxiu-rt']: oss_bucket = None oss_obj_path = { 'cs-rt': 'yiketu/static/fonts/', 'xiuxiu-rt': 'static/fonts/', }.get(oss_bucket, 'tmp/_fonts_minify/') parts = urlparse(font_url) path = parts.path base_name = os.path.basename(path) is_woff2 = re.search('(?i)woff2', base_name) oss_obj_path += 'woff2-ss/' if is_woff2 is True else 'woff-ss/' if oss_bucket != 'cs-rt': oss_obj_path += (obj_prefix.rstrip('/') + '/' if obj_prefix is not None else '') oss_obj_path += self.app_tool.uuid() + ('.woff2' if is_woff2 is True else '.woff') if with_txt is not None: text_file = os.path.join(self.base_dir, 'fonts_text/8000Hanzi.txt') print(text_file) args = [font_path, '--text-file=' + text_file, '--output-file=' + tmp_path] subset(args) ttf_path = tmp_path else: ttf_path = font_path font = TTFont(ttf_path) font.flavor = 'woff2' if is_woff2 is True else 'woff' font.save(tmp_path2) if os.path.exists(tmp_path): os.remove(tmp_path) if os.path.exists(tmp_path2): resp = self.app_tool.upload_to_oss_by_file(tmp_path2, oss_obj_path, bucket=oss_bucket) os.remove(tmp_path2) return resp def shortName(self, font): """Get the short name from the font's names table""" name = "" family = "" FONT_SPECIFIER_NAME_ID = 4 FONT_SPECIFIER_FAMILY_ID = 1 for record in font['name'].names: if b'\x00' in record.string: name_str = record.string.decode('utf-16-be') else: name_str = record.string.decode('utf-8') if record.nameID == FONT_SPECIFIER_NAME_ID and not name: name = name_str elif record.nameID == FONT_SPECIFIER_FAMILY_ID and not family: family = name_str if name and family: break return name, family def do_subset(self, font_url, text): font_path = self.download_font(font_url) tmp_path, filename = self.temp_path() tmp_subset_path = tmp_path + '.tff' subset_path = tmp_path + '.woff' args = [font_path, '--text=' + text, '--output-file=' + tmp_subset_path] subset(args) font = TTFont(tmp_subset_path) font.flavor = 'woff' font.save(subset_path) resp = self.app_tool.fail_result() if os.path.exists(tmp_subset_path): os.remove(tmp_subset_path) if os.path.exists(subset_path): object_path = os.path.join("tmp", "_xiuxiu_font_subsut", filename + '.woff') resp = self.app_tool.upload_to_oss_by_file(subset_path, object_path) os.remove(subset_path) return resp async def font_minify(self, request): form = await request.post() resp = self.app_tool.fail_result("data foramt error") if request.method == 'POST': font_url = None if 'font_url' not in form else form['font_url'] with_txt = None if 'with_txt' not in form else form['with_txt'] sign = None if 'sign' not in form else form['sign'] oss_bucket = None if 'oss_bucket' not in form else form['oss_bucket'] obj_prefix = None if 'obj_prefix' not in form else form['obj_prefix'] try: is_vaild = self.app_tool.verify_post_sign(sign, font_url) is_vaild = True if is_vaild is False or font_url is None: return web.Response(text=json.dumps(self.app_tool.fail_result("sign verify fail !"))) resp = self.do_font_minify(font_url, with_txt=with_txt, oss_bucket=oss_bucket, obj_prefix=obj_prefix) resp['with_txt'] = with_txt return web.Response(text=json.dumps(resp)) except ValueError as e: return web.Response(text=json.dumps(resp)) else: return web.Response(text=json.dumps(resp)) async def subset(self, request): form = await request.post() resp = self.app_tool.fail_result("格式解析失败") if request.method == 'POST': font_url = None if 'font_url' not in form else form['font_url'] text = None if 'text' not in form else form['text'] sign = None if 'sign' not in form else form['sign'] try: is_vaild = self.app_tool.verify_post_sign(sign, text) if is_vaild is False or font_url is None: return web.Response(text=json.dumps(self.app_tool.fail_result("非法的请求参数！"))) resp = self.do_subset(font_url, text) return web.Response(text=json.dumps(resp)) except ValueError as e: return web.Response(text=json.dumps(resp)) else: return web.Response(text=json.dumps(resp)) def serve_multiple(self, app, workers): sock = socket() sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) sock.bind(('0.0.0.0', LISTEN_PORT)) sock.set_inheritable(True) processes = [] for i in range(workers): process = Process(target=web.run_app, name=f'worker-{i}', kwargs=dict(app=app, sock=sock)) process.daemon = True process.start() processes.append(process) try: for process in processes: process.join() except KeyboardInterrupt: pass finally: for process in processes: process.terminate() sock.close() def main_multi(self, process_cnt): app = web.Application() app.router.add_get('/subset', self.subset, name="FontSubSet") app.router.add_post('/subset', self.subset, name="FontSubSet") app.router.add_get('/minify', self.font_minify, name="FontMinify") app.router.add_post('/minify', self.font_minify, name="FontMinify") cors = aiohttp_cors.setup(app, defaults={ "*": aiohttp_cors.ResourceOptions( allow_credentials=True, expose_headers="*", allow_headers="*", ) }) self.serve_multiple(app, process_cnt) def main(self): app = web.Application() app.router.add_get('/subset', self.subset, name="FontSubSet") app.router.add_post('/subset', self.subset, name="FontSubSet") app.router.add_get('/minify', self.font_minify, name="FontMinify") app.router.add_post('/minify', self.font_minify, name="FontMinify") web.run_app(app, port=LISTEN_PORT) if __name__ == '__main__': LISTEN_PORT = 8901 process_cnt = 1 FS = FontServer() FS.main_multi(process_cnt)