유투브 알고리즘을 따라해보자! 유사 유투바 알고리즘 만들기

RESUT API를 제공하는 웹서비스를 Resutful API라고 함.

 

- Rest :: Network상에서 Client와 Server 사이의 통신 방식 중 하나

- URI는 자원을 표현하는 데에 집중하고 행위에 대한 정의는 HTTP Method를 통해 하는 것이 REST한(CRUD) API를 설계하는 중심 규칙을 가지고 있음

    - CRUD Operation

       1. Create 

       2. Read

       3. Update

       4. Delete

       5. HEAD: Header 정보 조회

 

왜 REST가 필요한가?

    1. 다양한 클라이언트

    2. 애플리케이션 분리 및 통합

    3. 멀티 플랫폼 지원

 

REST 구성 요소

    1. 자원: Url, 클라이언트는 Url을 통해 자원을 지정하고 조작을 서버에 요청

    2. 행위: HTTP Method

    3. 표현: XML, JSON, TEXT 등 (json, xml이 일반적)

 

그냥 하나 만들어보자

플라스크를 이용했다.  

 

유투브 url을 넣어서 서버에 저장하는 코드를 작성했다.

특정 카테고리를 입력하면 해당 url 중 랜덤으로 하날 추천했다.

 

이름하야... 유사 유투바 알고리즘..

from flask import Flask
from flask_restx import reqparse, abort, Api, Resource
from collections import defaultdict
from pytube import YouTube
import random
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
 
app = Flask(__name__)
api = Api(
    app,
    version='0.1',
    title="유사 유투바 알고리즘",
    description="유투바의 알고리즘을 따라잡자!",
    terms_url="/youtube"
)
 
url_dic = defaultdict(dict)
 
parser = reqparse.RequestParser()
parser.add_argument('url')
parser.add_argument('category')
 
just_category_parser = reqparse.RequestParser()
just_category_parser.add_argument('category')
 
 
 
def abort_category(category):
    if category not in url_dic:
        abort(404, message="Category {} doesnt' exist".format(category))
 
 
def abort_url(category, url):
    if url not in url_dic[category]:
        abort(404, message="url {} doesnt' exist in Category {}".format(url, category))
 
 
@api.route('/youtube')
class Url(Resource):
    @api.expect(just_category_parser)
    def get(self):
        """ 지정된 카테고리의 url을 무작위로 추출합니다. """
        global url_dic
        args = just_category_parser.parse_args()
        category = args['category']
        abort_category(category=category)
        return random.choice(list(url_dic[category].values()))
 
 
    @api.expect(parser)
    def delete(self):
        """ 유투브 url을 삭제합니다. """
        global url_dic
        args = parser.parse_args()
        category = args['category']
        url = args['url']
        abort_url(category=category, url=url)
        del url_dic[category][url]
        return '', 204
 
    @api.expect(parser)
    def put(self):
        """ 유투브 url을 삽입합니다. """
        args = parser.parse_args()
        global url_dic
        category = args['category']
        url = args['url']
        yt_name = str(YouTube(url).title)
 
        if url in url_dic[category]:
            return '이미 추가하신 곡입니다', 202
        else:
 
            url_dic[args['category']][url] = yt_name
            return_value = url_dic[category][url]
            return return_value, 201
 
 
@api.route('/youtube/<string:category>')
class Url_list(Resource):
    def get(self, category):
        """ 특정 카테고리의 url을 모두 보여줍니다."""
        global url_dic
        return url_dic
 
@api.route('/yotube/all')
class Url_list_all(Resource):
    def get(self):
        """ 전체 url 목록을 보여줍니다. """
        global url_dic
        return url_dic
 
    
api.add_resource(Url, '/youtube')
api.add_resource(Url_list, '/youtube/<string:category>')
api.add_resource(Url_list, '/youtube/all')
 
if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=4444)

 

하지만 서버가 reset 될때마다 내 소중한 노동요 리스트들이 사라졌다..

db를 연결해주자.

 

database.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String
 
 
Base = declarative_base()
 
 
class url_list(Base):
    __tablename__ = 'url_list'
    url = Column(String(1000), primary_key=True)
    category = Column(String(100))
    url_name = Column(String(100))
 
    def __init__(self, url, category, url_name):
        self.url = url
        self.category = category
        self.url_name = url_name
 
def insert_url(session, url, url_name, category):
    try:
        new_info = url_list(url=url, category=category ,url_name=url_name)
        session.add(new_info)
        session.commit()
 
    except Exception as e:
        print("\nUnique value insert exception\n")
        session.rollback()
 
 
def delete_url(session, url):
    deleted_objects = url_list.__table__.delete().where(url_list.url.is_(url))
    session.execute(deleted_objects)
    session.commit()
 
 
def select_url(session, category=None):
    if category is not None: #전체 다 보여
        urls = session.query(url_list.url_name, url_list.url)\
        .filter(url_list.category == category).all()
 
    else:
        urls = session.query(url_list.category, url_list.url_name, url_list.url).all()
 
    return urls

app.py

import sqlalchemy
from flask import Flask
from flask_restx import reqparse, abort, Api, Resource
from collections import defaultdict
from pytube import YouTube
from sqlalchemy.orm import sessionmaker
 
import database as db
import random
import ssl
 
ssl._create_default_https_context = ssl._create_unverified_context
 
app = Flask(__name__)
api = Api(
    app,
    version='0.1',
    title="유사 유투바 알고리즘",
    description="유투바의 알고리즘을 따라잡자!",
    terms_url="/youtube"
)
 
 
url_dic = defaultdict(dict)
 
parser = reqparse.RequestParser()
parser.add_argument('url')
parser.add_argument('category')
 
just_category_parser = reqparse.RequestParser()
just_category_parser.add_argument('category')
 
just_url_parser = reqparse.RequestParser()
just_url_parser.add_argument('url')
 
 
@api.route('/youtube')
class Url(Resource):
    @api.expect(just_category_parser)
    def get(self):
        """ 지정된 카테고리의 url을 무작위로 추출합니다. """
        global url_dic
        args = just_category_parser.parse_args()
        category = args['category']
 
        category_url_list = db.select_url(session=dbconnect(), category=category)
        if category_url_list:
            return random.choice(list(category_url_list)),201
        else:
            return "해당 카테고리에 url이 존재하지 않습니다!"
 
 
    @api.expect(just_url_parser)
    def delete(self):
        """ 유투브 url을 삭제합니다. """
        global url_dic
        args = just_url_parser.parse_args()
        url = args['url']
        db.delete_url(session=dbconnect(),url=url)
        return f'{url} 삭제 완료!', 201
 
    @api.expect(parser)
    def put(self):
        """ 유투브 url을 삽입합니다. """
        args = parser.parse_args()
        global url_dic
        category = args['category']
        url = args['url']
        yt_name = str(YouTube(url).title)
 
        if url in url_dic[category]:
            return '이미 추가하신 곡입니다'
        else:
            db.insert_url(session=dbconnect(), url=url, category=category, url_name=yt_name)
 
            return_value = "{}:{}".format(yt_name, url)
            return return_value, 201
 
 
@api.route('/youtube/<string:category>')
class Url_list(Resource):
    def get(self, category):
        """ 특정 카테고리의 url을 모두 보여줍니다."""
        return db.select_url(session=dbconnect(), category=category), 201
 
 
@api.route('/youtube/all')
class Url_list_all(Resource):
    def get(self):
        """ 전체 url 목록을 보여줍니다. """
        return db.select_url(session=dbconnect()), 201
 
 
api.add_resource(Url, '/youtube')
api.add_resource(Url_list, '/youtube/<string:category>')
api.add_resource(Url_list_all, '/youtube/all')
 
 
def dbconnect():
    engine = sqlalchemy.create_engine('sqlite:///youtube.db')
    Session = sessionmaker(bind=engine)
    return Session()
 
 
if __name__ == '__main__':
    engine = sqlalchemy.create_engine('sqlite:///youtube.db')
    db.Base.metadata.create_all(engine)
    app.run(debug=True, host='0.0.0.0', port=1919)

완성본은 아래와 같음.

내가 디비에 put한 데이터들 중에서 

랜덤으로 음악 url을 추천해주도록 설계되어있음.

아래처럼 간단하게 웹 붙여서 요즘 필요할 떄 듣는다.. ㅎㅎ

 

Reference

gmlwjd9405.github.io/2018/09/21/rest-and-restful.html

niceman.tistory.com/101