Выдача наград за голосование

В этом обновлении убран кик при голосовании, теперь скрипт будет дожидаться выхода из игры, чтобы выдать бонус, ну и добавлена расшифровка кодов, теперь можно дебажить скрипт.

Пример выдачи призов за голосовалку на l2top не реклама, только его и знаю.
Понадобится таблица куда будем сгружать полученные парсером даты и имена чаров.
Будет она в базе lin2db

Спойлер: dbo.votes

SQL:

USE [lin2db]
GO

/****** Object:  Table [dbo].[votes]    Script Date: 07/02/2024 08:33:22 ******/
SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO

CREATE TABLE [dbo].[votes](
    [Id] [int] IDENTITY(1,1) NOT NULL,
    [Date] [datetime] NULL,
    [char_name] [nvarchar](255) NULL,
    [BonusIssued] [bit] NULL,
PRIMARY KEY CLUSTERED
(
    [Id] ASC
)WITH (PAD_INDEX  = OFF, STATISTICS_NORECOMPUTE  = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS  = ON, ALLOW_PAGE_LOCKS  = ON) ON [PRIMARY]
) ON [PRIMARY]

GO

ALTER TABLE [dbo].[votes] ADD  CONSTRAINT [DF_BonusIssued]  DEFAULT ((0)) FOR [BonusIssued]
GO

Далее понадобится создать четыре процедуры в этой же базе.
Процедура создания записей в таблице на основе данных полученных из парсера.

Спойлер: InsertVote

SQL:

USE [lin2db]
GO

/****** Object:  StoredProcedure [dbo].[InsertVote]    Script Date: 07/02/2024 08:52:46 ******/
SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO


CREATE PROCEDURE [dbo].[InsertVote]
    @Date DATETIME,
    @char_name NVARCHAR(255)
AS
BEGIN
    IF NOT EXISTS (SELECT 1 FROM Votes WHERE Date = @Date AND char_name = @char_name)
    BEGIN
        INSERT INTO Votes (Date, char_name, BonusIssued)
        VALUES (@Date, @char_name, 0);
    END
END;
GO

Процедура получения имен чаров и даты для выдачи бонуса.

Спойлер: GetPendingBonusCharNames

SQL:

USE [lin2db]
GO

/****** Object:  StoredProcedure [dbo].[GetPendingBonusCharNames]    Script Date: 07/02/2024 08:36:28 ******/
SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO


CREATE PROCEDURE [dbo].[GetPendingBonusCharNames]
AS
BEGIN
    SELECT date, char_name
    FROM votes
    WHERE BonusIssued = 0;
END;

GO

Процедура получения ID чара по его имени.

Спойлер: GetCharIdByCharName

Python:

USE [lin2db]
GO

/****** Object:  StoredProcedure [dbo].[GetCharIdByCharName]    Script Date: 07/02/2024 15:56:28 ******/
SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO


CREATE PROCEDURE [dbo].[GetCharIdByCharName]
    @char_name NVARCHAR(255)
AS
BEGIN
    DECLARE @char_id INT;

    -- Получаем char_id из базы данных lin2world
    SELECT @char_id = char_id
    FROM lin2world.dbo.user_data
    WHERE char_name = @char_name;

    -- Возвращаем найденный char_id
    SELECT @char_id AS char_id;
END;
GO

Процедура выдачи бонуса.

Спойлер: GiveBonus

SQL:

USE [lin2db]
GO

/****** Object:  StoredProcedure [dbo].[GiveBonus]    Script Date: 07/02/2024 09:14:14 ******/
SET ANSI_NULLS ON
GO

SET QUOTED_IDENTIFIER ON
GO


CREATE PROCEDURE [dbo].[GiveBonus]
    @Date DATETIME,
    @char_name NVARCHAR(255)
AS
BEGIN
    UPDATE Votes
    SET BonusIssued = 1
    WHERE char_name = @char_name AND Date = @Date;

END;


GO

Теперь сама программа, написал её на python т.к. изучаю его.

Спойлер: cached.py

Python:

import socket
import struct

class CacheD:
    def __init__(self, cached_ip, cached_port, webadmin='Admin'):
        self.cached_ip = cached_ip
        self.cached_port = cached_port
        self.webadmin = webadmin
        self.connected = False
        self.fsockerror = 'Unable to connect with CacheD'
        self.socketerrors = {
            "1": "OK.",
            "01": "Error.",
            "02": "Arguments need.",
            "03": "Arguments invalid.",
            "04": "Char not found.",
            "05": "Warehouse not found.",
            "06": "Account not found.",
            "07": "Char in game.",
            "08": "Too many chars.",
            "09": "Char in pledge.",
            "010": "Char pledge owner.",
            "011": "Cannot ban.",
            "012": "Name exist.",
            "013": "Obsolete.",
            "014": "Invalid char name.",
            "015": "Char not in game.",
            "016": "Same char.",
            "017": "Char not in pledge.",
            "018": "Char pledge master.",
            "019": "Server not connected.",
            "020": "Create pet failed.",
            "021": "Pledge exist.",
            "022": "No chars.",
            "023": "Invalid announce id.",
            "024": "Pledge not found.",
            "025": "Castle not found.",
            "026": "Pet not found."
        }
    def tounicode(self, string):
        rs = ""
        for char in string:
            rs += char + '\x00'
        rs += '\x00\x00'
        return rs.encode('utf-16le')
    def CacheDInteractive(self, buf):
        try:
            with socket.create_connection((self.cached_ip, self.cached_port),
                                          timeout=5) as sock:
                self.connected = True
                packet_length = struct.pack("<h", len(buf) + 2)
                packet = packet_length + buf
                sock.sendall(packet)
                length_data = sock.recv(2)
                length = struct.unpack("<h", length_data)[0]
                response_id_data = sock.recv(1)
                response_id = struct.unpack("<b", response_id_data)[0]
                remaining_length = length - 3
                response_data = sock.recv(remaining_length)
                rs = ''
                for i in range(0, len(response_data), 4):
                    read_data = response_data[i:i + 4]
                    if len(read_data) < 4:
                        break
                    read = struct.unpack("<i", read_data)[0]
                    rs += str(read)
                result = self.socketerrors.get(rs, "Unknown error.")
                return result
        except socket.error as e:
            self.connected = False
            return self.fsockerror
    def AddItem2Packet(self, char_id, warehouse, item_type, q, enchant, eroded,
                       bless, wished):
        bless = 0
        q_h = int(q / 0x100000000)
        if q_h < 0:
            q_h = 0
        else:
            q = q - (q_h * 0x100000000)
        buf = struct.pack("B", 55)
        buf += struct.pack("I", char_id)
        buf += struct.pack("I", warehouse)
        buf += struct.pack("I", item_type)
        buf += struct.pack("I", q)
        buf += struct.pack("I", q_h)
        buf += struct.pack("I", enchant)
        buf += struct.pack("I", eroded)
        buf += struct.pack("I", bless)
        buf += struct.pack("I", wished)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += struct.pack("I", 0)
        buf += self.tounicode(self.webadmin)
        return self.CacheDInteractive(buf)
    def CheckCharacterPacket(self, char_id):
        buf = struct.pack("<B", 1) + struct.pack(
            "<I", char_id) + self.tounicode(self.webadmin)
        return self.CacheDInteractive(buf)
    def KickCharacterPacket(self, char_id):
        buf = struct.pack("<B", 5) + struct.pack(
            "<I", char_id) + self.tounicode(self.webadmin)
        return self.CacheDInteractive(buf)

Спойлер: vote_parser.py

Python:

import requests
from bs4 import BeautifulSoup
import pyodbc

def parse_and_insert_votes(connection_string, url):
    conn = pyodbc.connect(connection_string)
    cursor = conn.cursor()
    response = requests.get(url)
    response.encoding = 'utf-8'
    soup = BeautifulSoup(response.text, 'html.parser')
    text = soup.get_text()
    lines = [line for line in text.strip().split('\n') if line]
    data = [line.split('\t') for line in lines]
    for row in data[1:]:
        if len(row) > 0:
            date_time = row[0]
            char_name = row[1]
            if 'http' not in date_time:
                cursor.execute("EXEC InsertVote @Date = ?, @char_name = ?",
                               date_time, char_name)
    conn.commit()
    cursor.close()
    conn.close()

Спойлер: main.py

Python:

import pyodbc
import time
from cached import CacheD
from vote_parser import parse_and_insert_votes
# Параметры подключения к MSSQL
server = '127.0.0.1'
database = 'lin2db'
username = 'admin'
password = 'admin'
driver = 'ODBC Driver 17 for SQL Server'
connection_string = (
    f'DRIVER={driver};SERVER={server};DATABASE={database};UID={username};PWD={password}'
)
# URL для парсинга
url = 'https://l2top.ru/editServ/?adminAct=lastVotes&uid=ВАШ_UID&key=ВАШ_KEY'
# Парсинг и вставка данных
parse_and_insert_votes(connection_string, url)
conn = pyodbc.connect(connection_string)
cursor = conn.cursor()
# Получаем char_id для каждого char_name и выдаем бонус
cached_ip = '127.0.0.1'
cached_port = 2012
item_id = 57
item_count = 1
cacheD = CacheD(cached_ip, cached_port)
# Использование процедуры для получения всех char_name, где BonusIssued = 0
cursor.execute("EXEC GetPendingBonusCharNames")
char_names = cursor.fetchall()
for row in char_names:
    date = row[0]
    char_name = row[1]
    cursor.execute("EXEC GetCharIdByCharName @char_name = ?", char_name)
    char_id_row = cursor.fetchone()
    if char_id_row is None or char_id_row[0] is None:
        cursor.execute("EXEC GiveBonus @Date = ?, @char_name = ?", date,
                       char_name)
        continue
    else:
        char_id = char_id_row[0]
        if cacheD.CheckCharacterPacket(char_id) == "OK.":
            continue
        else:
            cacheD.AddItem2Packet(char_id, 0, item_id, item_count, 0, 0, 0, 0)
            # Обновляем поле BonusIssued
            cursor.execute("EXEC GiveBonus @Date = ?, @char_name = ?", date,
                           char_name)
conn.commit()
cursor.close()
conn.close()