短链生成器 - redis实现

短链生成器 - redis实现

黄鹏宇 1,206 2021-12-05

实现的功能

  • 生成短链
  • 跳转
  • 限制生成频率

使用的技术栈

  • python
  • flask
  • flask_limiter
  • redis

效果

hpy.im

跳转

代码

下载链接
https://cdns.qdu.life/短链接生成器代码.rar

requirements.txt

pipreqs . --encoding=utf8 --force

Flask==2.0.1
Flask_Limiter==2.0.2
gevent==21.8.0
redis==4.0.2

short_server.py

# -*- coding:utf8 -*-

import time
import json
import os
import sys
import threading
import redis
from redis import Redis

from short_url import ShortyUrl
from flask import Flask,jsonify,request,render_template,redirect
from gevent import pywsgi

# 限制访问速度
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

LIMITS_REDIS_STORAGE = "redis://localhost:6379"

app = Flask(__name__)

# 根据访问者的IP记录访问次数
# 注意:在真实开发中,大部分项目都配置了Nginx,如果直接使用get_remote_address,获取到的是Nginx服务器的地址,相当于来自该Nginx服务器的所有请求会被当做同一个IP访问,所以项目中一般都是自定义key_func

limiter = Limiter(
    app,
    key_func=limit_key_func,   # 根据访问者的IP记录访问次数
    default_limits=["200/day", "50/hour" , "10/minute"],  # 默认限制,一天最多访问200次,一小时最多访问50次
    storage_uri = LIMITS_REDIS_STORAGE
)

def limit_key_func():
	return str(request.headers.get("X-Forwarded-For", '127.0.0.1'))
	
@app.route('/s')
def generateUrl():
	url = request.args.get("url")
	if(url != None and url != ""):
		return "hpy.im/" +  str(shortly_url.shorten(url))
	else:
		return "err"

@app.route("/")
@limiter.exempt # 不限制
def html():
	return render_template('index.html')

@app.route("/<path:path>")
@limiter.exempt # 不限制
def navigator(path):
	url = shortly_url.restore(path)
	if(url == None):
		return '无此短链'
	else:
		url = str(url,encoding = "utf8")
		# 判断有没有请求头
		if(not url.startswith("http")):
			url = "https://" + url 
		return redirect(url)

if __name__ == '__main__':
	pool = redis.ConnectionPool(host='127.0.0.1')   #实现一个连接池
	client = Redis(decode_responses = True,connection_pool=pool)
	shortly_url = ShortyUrl(client)	
	server = pywsgi.WSGIServer(('0.0.0.0', 5008), app)
	server.serve_forever()

short_url.py

from base36 import base10_to_base36

ID_COUNTER = "ShortlyUrl::id_counter"
URL_HASH = "ShortlyUrl::url_hash"

class ShortyUrl:
	def __init__(self,client):
		self.client = client

	def shorten(self,target_url):
		new_id = self.client.incr(ID_COUNTER)
		short_id = base10_to_base36(new_id)
		self.client.hset(URL_HASH,short_id,target_url)
		return short_id

	def restore(self,short_id):
		return self.client.hget(URL_HASH,short_id)

base36.py

def base10_to_base36(number):
	alphabets = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
	result = ""
	while number != 0:
		number,i = divmod(number,36)
		result = (alphabets[i] + result)
	return result or alphabets[0]

templates/index.html

<!DOCTYPE html>
<html>

<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>短链生成器</title>
    <link rel="stylesheet" href="https://qidian.gtimg.com/lulu/edge/css/common/ui/Button.css">
    <link rel="stylesheet" href="https://qidian.gtimg.com/lulu/edge/css/common/ui/Input.css">
    <link rel="stylesheet" href="https://qidian.gtimg.com/lulu/edge/css/common/ui/Placeholder.css">
    <link rel="stylesheet" href="https://qidian.gtimg.com/lulu/edge/css/common/ui/Loading.css">
    <link rel="stylesheet" href="https://qidian.gtimg.com/lulu/edge/css/common/ui/LightTip.css">
</head>

<body style="overflow: hidden;">
    <div style="text-align: center;padding:30px;padding-top:10vh;">
        <h1 style="padding-bottom:3vh">短链生成器(hpy.im)</h1>
        <input id="originUrl" type="text" class="ui-input" name="url" width="100%" placeholder="在此输入你的网址">
        <div style="padding-top:2vh" style="cursor: pointer; font-size:0">
            <div class="ui-input-x" id="resultBtn">
                <input id="result" value="" class="ui-input" maxlength="200" placeholder="生成结果">
                <label class="ui-input-count">
                    <output>点击复制</span>
                </label>
            </div>
        </div>
        <div style="padding-top:2vh;padding-bottom:2vh">
            <button data-type="primary" role="button" id="btn" class="ui-button" width="100%">缩短</button>
        </div>
    </div>
    <div style="color:lightgrey;text-align:center;bottom: 0;position: absolute;width: 97%;">
        <h5><a style="color:grey;font-weight: normal;" href="https://github.com/suyu610/nav">项目代码</a></h5>
        <h5>生成频率: 200/day | 50/hour | 10/min </h5>
    </div>
</body>
<script type="text/javascript" src="https://cdns.qdu.life/jiayao/static/js/jquery-3.6.0.min.js"></script>
<script type="module" src="https://qidian.gtimg.com/lulu/edge/js/common/ui/Loading.js"></script>
<script type="module" src="https://qidian.gtimg.com/lulu/edge/js/common/ui/LightTip.js"></script>
<script type="text/javascript">
$('#resultBtn').click(function() {
    var copyCon = document.getElementById("result");
    if (copyCon.value == "") {
        return;
    }
    copyCon.select(); // 选择对象
    document.execCommand("Copy"); // 执行浏览器复制命令
    alert("已复制");
})



var button = document.getElementById("btn")
button.addEventListener('click', function() {
    button.disabled = true;
    button.loading = true;

    var originUrl = document.getElementById("originUrl")
    if (originUrl.value == '') {
        alert("网址不能为空")
        button.disabled = false;
        button.loading = false;
        return;
    }

    $.ajax({
        type: "GET",
        url: "https://www.hpy.im/s?url=" + originUrl.value,
        success: function(data) {
            button.disabled = false;
            button.loading = false;
            $("#result").val(data)
            new LightTip('生成成功', 1000, 'success');

            console.log(data)
        },
        error: function(e) {
            if (e.status == '429') {
                new LightTip('生成频率过快', 1000, 'error');
            } else {
                new LightTip('发生未知错误', 2000, 'error');
            }
            button.disabled = false;
            button.loading = false;
        }
    });
});
</script>

</html>